141init_environment(void) 142{ 143 struct hio *hio; 144 unsigned int ii; 145 146 /* 147 * Initialize lists, their locks and theirs condition variables. 148 */ 149 TAILQ_INIT(&hio_free_list); 150 mtx_init(&hio_free_list_lock); 151 cv_init(&hio_free_list_cond); 152 TAILQ_INIT(&hio_disk_list); 153 mtx_init(&hio_disk_list_lock); 154 cv_init(&hio_disk_list_cond); 155 TAILQ_INIT(&hio_send_list); 156 mtx_init(&hio_send_list_lock); 157 cv_init(&hio_send_list_cond); 158 159 /* 160 * Allocate requests pool and initialize requests. 161 */ 162 for (ii = 0; ii < HAST_HIO_MAX; ii++) { 163 hio = malloc(sizeof(*hio)); 164 if (hio == NULL) { 165 pjdlog_exitx(EX_TEMPFAIL, 166 "Unable to allocate memory (%zu bytes) for hio request.", 167 sizeof(*hio)); 168 } 169 hio->hio_data = malloc(MAXPHYS); 170 if (hio->hio_data == NULL) { 171 pjdlog_exitx(EX_TEMPFAIL, 172 "Unable to allocate memory (%zu bytes) for gctl_data.", 173 (size_t)MAXPHYS); 174 } 175 hio_clear(hio); 176 TAILQ_INSERT_HEAD(&hio_free_list, hio, hio_next); 177 } 178} 179 180static void 181init_local(struct hast_resource *res) 182{ 183 184 if (metadata_read(res, true) == -1) 185 exit(EX_NOINPUT); 186} 187 188static void 189init_remote(struct hast_resource *res, struct nv *nvin) 190{ 191 uint64_t resuid; 192 struct nv *nvout; 193 unsigned char *map; 194 size_t mapsize; 195 196#ifdef notyet 197 /* Setup direction. */ 198 if (proto_send(res->hr_remoteout, NULL, 0) == -1) 199 pjdlog_errno(LOG_WARNING, "Unable to set connection direction"); 200#endif 201 202 nvout = nv_alloc(); 203 nv_add_int64(nvout, (int64_t)res->hr_datasize, "datasize"); 204 nv_add_int32(nvout, (int32_t)res->hr_extentsize, "extentsize"); 205 resuid = nv_get_uint64(nvin, "resuid"); 206 res->hr_primary_localcnt = nv_get_uint64(nvin, "localcnt"); 207 res->hr_primary_remotecnt = nv_get_uint64(nvin, "remotecnt"); 208 nv_add_uint64(nvout, res->hr_secondary_localcnt, "localcnt"); 209 nv_add_uint64(nvout, res->hr_secondary_remotecnt, "remotecnt"); 210 mapsize = activemap_calc_ondisk_size(res->hr_local_mediasize - 211 METADATA_SIZE, res->hr_extentsize, res->hr_local_sectorsize); 212 map = malloc(mapsize); 213 if (map == NULL) { 214 pjdlog_exitx(EX_TEMPFAIL, 215 "Unable to allocate memory (%zu bytes) for activemap.", 216 mapsize); 217 } 218 /* 219 * When we work as primary and secondary is missing we will increase 220 * localcnt in our metadata. When secondary is connected and synced 221 * we make localcnt be equal to remotecnt, which means nodes are more 222 * or less in sync. 223 * Split-brain condition is when both nodes are not able to communicate 224 * and are both configured as primary nodes. In turn, they can both 225 * make incompatible changes to the data and we have to detect that. 226 * Under split-brain condition we will increase our localcnt on first 227 * write and remote node will increase its localcnt on first write. 228 * When we connect we can see that primary's localcnt is greater than 229 * our remotecnt (primary was modified while we weren't watching) and 230 * our localcnt is greater than primary's remotecnt (we were modified 231 * while primary wasn't watching). 232 * There are many possible combinations which are all gathered below. 233 * Don't pay too much attention to exact numbers, the more important 234 * is to compare them. We compare secondary's local with primary's 235 * remote and secondary's remote with primary's local. 236 * Note that every case where primary's localcnt is smaller than 237 * secondary's remotecnt and where secondary's localcnt is smaller than 238 * primary's remotecnt should be impossible in practise. We will perform 239 * full synchronization then. Those cases are marked with an asterisk. 240 * Regular synchronization means that only extents marked as dirty are 241 * synchronized (regular synchronization). 242 * 243 * SECONDARY METADATA PRIMARY METADATA 244 * local=3 remote=3 local=2 remote=2* ?! Full sync from secondary. 245 * local=3 remote=3 local=2 remote=3* ?! Full sync from primary. 246 * local=3 remote=3 local=2 remote=4* ?! Full sync from primary. 247 * local=3 remote=3 local=3 remote=2 Primary is out-of-date, 248 * regular sync from secondary. 249 * local=3 remote=3 local=3 remote=3 Regular sync just in case. 250 * local=3 remote=3 local=3 remote=4* ?! Full sync from primary. 251 * local=3 remote=3 local=4 remote=2 Split-brain condition. 252 * local=3 remote=3 local=4 remote=3 Secondary out-of-date, 253 * regular sync from primary. 254 * local=3 remote=3 local=4 remote=4* ?! Full sync from primary. 255 */ 256 if (res->hr_resuid == 0) { 257 /* 258 * Provider is used for the first time. If primary node done no 259 * writes yet as well (we will find "virgin" argument) then 260 * there is no need to synchronize anything. If primary node 261 * done any writes already we have to synchronize everything. 262 */ 263 PJDLOG_ASSERT(res->hr_secondary_localcnt == 0); 264 res->hr_resuid = resuid; 265 if (metadata_write(res) == -1) 266 exit(EX_NOINPUT); 267 if (nv_exists(nvin, "virgin")) { 268 free(map); 269 map = NULL; 270 mapsize = 0; 271 } else { 272 memset(map, 0xff, mapsize); 273 } 274 nv_add_int8(nvout, 1, "virgin"); 275 nv_add_uint8(nvout, HAST_SYNCSRC_PRIMARY, "syncsrc"); 276 } else if (res->hr_resuid != resuid) { 277 char errmsg[256]; 278 279 free(map); 280 (void)snprintf(errmsg, sizeof(errmsg), 281 "Resource unique ID mismatch (primary=%ju, secondary=%ju).", 282 (uintmax_t)resuid, (uintmax_t)res->hr_resuid); 283 pjdlog_error("%s", errmsg); 284 nv_add_string(nvout, errmsg, "errmsg"); 285 if (hast_proto_send(res, res->hr_remotein, nvout, 286 NULL, 0) == -1) { 287 pjdlog_exit(EX_TEMPFAIL, 288 "Unable to send response to %s", 289 res->hr_remoteaddr); 290 } 291 nv_free(nvout); 292 exit(EX_CONFIG); 293 } else if ( 294 /* Is primary out-of-date? */ 295 (res->hr_secondary_localcnt > res->hr_primary_remotecnt && 296 res->hr_secondary_remotecnt == res->hr_primary_localcnt) || 297 /* Are the nodes more or less in sync? */ 298 (res->hr_secondary_localcnt == res->hr_primary_remotecnt && 299 res->hr_secondary_remotecnt == res->hr_primary_localcnt) || 300 /* Is secondary out-of-date? */ 301 (res->hr_secondary_localcnt == res->hr_primary_remotecnt && 302 res->hr_secondary_remotecnt < res->hr_primary_localcnt)) { 303 /* 304 * Nodes are more or less in sync or one of the nodes is 305 * out-of-date. 306 * It doesn't matter at this point which one, we just have to 307 * send out local bitmap to the remote node. 308 */ 309 if (pread(res->hr_localfd, map, mapsize, METADATA_SIZE) != 310 (ssize_t)mapsize) { 311 pjdlog_exit(LOG_ERR, "Unable to read activemap"); 312 } 313 if (res->hr_secondary_localcnt > res->hr_primary_remotecnt && 314 res->hr_secondary_remotecnt == res->hr_primary_localcnt) { 315 /* Primary is out-of-date, sync from secondary. */ 316 nv_add_uint8(nvout, HAST_SYNCSRC_SECONDARY, "syncsrc"); 317 } else { 318 /* 319 * Secondary is out-of-date or counts match. 320 * Sync from primary. 321 */ 322 nv_add_uint8(nvout, HAST_SYNCSRC_PRIMARY, "syncsrc"); 323 } 324 } else if (res->hr_secondary_localcnt > res->hr_primary_remotecnt && 325 res->hr_primary_localcnt > res->hr_secondary_remotecnt) { 326 /* 327 * Not good, we have split-brain condition. 328 */ 329 free(map); 330 pjdlog_error("Split-brain detected, exiting."); 331 nv_add_string(nvout, "Split-brain condition!", "errmsg"); 332 if (hast_proto_send(res, res->hr_remotein, nvout, 333 NULL, 0) == -1) { 334 pjdlog_exit(EX_TEMPFAIL, 335 "Unable to send response to %s", 336 res->hr_remoteaddr); 337 } 338 nv_free(nvout); 339 /* Exit on split-brain. */ 340 event_send(res, EVENT_SPLITBRAIN); 341 exit(EX_CONFIG); 342 } else /* if (res->hr_secondary_localcnt < res->hr_primary_remotecnt || 343 res->hr_primary_localcnt < res->hr_secondary_remotecnt) */ { 344 /* 345 * This should never happen in practise, but we will perform 346 * full synchronization. 347 */ 348 PJDLOG_ASSERT(res->hr_secondary_localcnt < res->hr_primary_remotecnt || 349 res->hr_primary_localcnt < res->hr_secondary_remotecnt); 350 mapsize = activemap_calc_ondisk_size(res->hr_local_mediasize - 351 METADATA_SIZE, res->hr_extentsize, 352 res->hr_local_sectorsize); 353 memset(map, 0xff, mapsize); 354 if (res->hr_secondary_localcnt > res->hr_primary_remotecnt) { 355 /* In this one of five cases sync from secondary. */ 356 nv_add_uint8(nvout, HAST_SYNCSRC_SECONDARY, "syncsrc"); 357 } else { 358 /* For the rest four cases sync from primary. */ 359 nv_add_uint8(nvout, HAST_SYNCSRC_PRIMARY, "syncsrc"); 360 } 361 pjdlog_warning("This should never happen, asking for full synchronization (primary(local=%ju, remote=%ju), secondary(local=%ju, remote=%ju)).", 362 (uintmax_t)res->hr_primary_localcnt, 363 (uintmax_t)res->hr_primary_remotecnt, 364 (uintmax_t)res->hr_secondary_localcnt, 365 (uintmax_t)res->hr_secondary_remotecnt); 366 } 367 nv_add_uint32(nvout, (uint32_t)mapsize, "mapsize"); 368 if (hast_proto_send(res, res->hr_remotein, nvout, map, mapsize) == -1) { 369 pjdlog_exit(EX_TEMPFAIL, "Unable to send activemap to %s", 370 res->hr_remoteaddr); 371 } 372 if (map != NULL) 373 free(map); 374 nv_free(nvout); 375#ifdef notyet 376 /* Setup direction. */ 377 if (proto_recv(res->hr_remotein, NULL, 0) == -1) 378 pjdlog_errno(LOG_WARNING, "Unable to set connection direction"); 379#endif 380} 381 382void 383hastd_secondary(struct hast_resource *res, struct nv *nvin) 384{ 385 sigset_t mask; 386 pthread_t td; 387 pid_t pid; 388 int error, mode, debuglevel; 389 390 /* 391 * Create communication channel between parent and child. 392 */ 393 if (proto_client(NULL, "socketpair://", &res->hr_ctrl) == -1) { 394 KEEP_ERRNO((void)pidfile_remove(pfh)); 395 pjdlog_exit(EX_OSERR, 396 "Unable to create control sockets between parent and child"); 397 } 398 /* 399 * Create communication channel between child and parent. 400 */ 401 if (proto_client(NULL, "socketpair://", &res->hr_event) == -1) { 402 KEEP_ERRNO((void)pidfile_remove(pfh)); 403 pjdlog_exit(EX_OSERR, 404 "Unable to create event sockets between child and parent"); 405 } 406 407 pid = fork(); 408 if (pid == -1) { 409 KEEP_ERRNO((void)pidfile_remove(pfh)); 410 pjdlog_exit(EX_OSERR, "Unable to fork"); 411 } 412 413 if (pid > 0) { 414 /* This is parent. */ 415 proto_close(res->hr_remotein); 416 res->hr_remotein = NULL; 417 proto_close(res->hr_remoteout); 418 res->hr_remoteout = NULL; 419 /* Declare that we are receiver. */ 420 proto_recv(res->hr_event, NULL, 0); 421 /* Declare that we are sender. */ 422 proto_send(res->hr_ctrl, NULL, 0); 423 res->hr_workerpid = pid; 424 return; 425 } 426 427 gres = res; 428 mode = pjdlog_mode_get(); 429 debuglevel = pjdlog_debug_get(); 430 431 /* Declare that we are sender. */ 432 proto_send(res->hr_event, NULL, 0); 433 /* Declare that we are receiver. */ 434 proto_recv(res->hr_ctrl, NULL, 0); 435 descriptors_cleanup(res); 436 437 descriptors_assert(res, mode); 438 439 pjdlog_init(mode); 440 pjdlog_debug_set(debuglevel); 441 pjdlog_prefix_set("[%s] (%s) ", res->hr_name, role2str(res->hr_role)); 442 setproctitle("%s (%s)", res->hr_name, role2str(res->hr_role)); 443 444 PJDLOG_VERIFY(sigemptyset(&mask) == 0); 445 PJDLOG_VERIFY(sigprocmask(SIG_SETMASK, &mask, NULL) == 0); 446 447 /* Error in setting timeout is not critical, but why should it fail? */ 448 if (proto_timeout(res->hr_remotein, 2 * HAST_KEEPALIVE) == -1) 449 pjdlog_errno(LOG_WARNING, "Unable to set connection timeout"); 450 if (proto_timeout(res->hr_remoteout, res->hr_timeout) == -1) 451 pjdlog_errno(LOG_WARNING, "Unable to set connection timeout"); 452 453 init_local(res); 454 init_environment(); 455 456 if (drop_privs(res) != 0) 457 exit(EX_CONFIG); 458 pjdlog_info("Privileges successfully dropped."); 459 460 /* 461 * Create the control thread before sending any event to the parent, 462 * as we can deadlock when parent sends control request to worker, 463 * but worker has no control thread started yet, so parent waits. 464 * In the meantime worker sends an event to the parent, but parent 465 * is unable to handle the event, because it waits for control 466 * request response. 467 */ 468 error = pthread_create(&td, NULL, ctrl_thread, res); 469 PJDLOG_ASSERT(error == 0); 470 471 init_remote(res, nvin); 472 event_send(res, EVENT_CONNECT); 473 474 error = pthread_create(&td, NULL, recv_thread, res); 475 PJDLOG_ASSERT(error == 0); 476 error = pthread_create(&td, NULL, disk_thread, res); 477 PJDLOG_ASSERT(error == 0); 478 (void)send_thread(res); 479} 480 481static void 482reqlog(int loglevel, int debuglevel, int error, struct hio *hio, 483 const char *fmt, ...) 484{ 485 char msg[1024]; 486 va_list ap; 487 int len; 488 489 va_start(ap, fmt); 490 len = vsnprintf(msg, sizeof(msg), fmt, ap); 491 va_end(ap); 492 if ((size_t)len < sizeof(msg)) { 493 switch (hio->hio_cmd) { 494 case HIO_READ: 495 (void)snprintf(msg + len, sizeof(msg) - len, 496 "READ(%ju, %ju).", (uintmax_t)hio->hio_offset, 497 (uintmax_t)hio->hio_length); 498 break; 499 case HIO_DELETE: 500 (void)snprintf(msg + len, sizeof(msg) - len, 501 "DELETE(%ju, %ju).", (uintmax_t)hio->hio_offset, 502 (uintmax_t)hio->hio_length); 503 break; 504 case HIO_FLUSH: 505 (void)snprintf(msg + len, sizeof(msg) - len, "FLUSH."); 506 break; 507 case HIO_WRITE: 508 (void)snprintf(msg + len, sizeof(msg) - len, 509 "WRITE(%ju, %ju).", (uintmax_t)hio->hio_offset, 510 (uintmax_t)hio->hio_length); 511 break; 512 case HIO_KEEPALIVE: 513 (void)snprintf(msg + len, sizeof(msg) - len, "KEEPALIVE."); 514 break; 515 default: 516 (void)snprintf(msg + len, sizeof(msg) - len, 517 "UNKNOWN(%u).", (unsigned int)hio->hio_cmd); 518 break; 519 } 520 } 521 pjdlog_common(loglevel, debuglevel, error, "%s", msg); 522} 523 524static int 525requnpack(struct hast_resource *res, struct hio *hio, struct nv *nv) 526{ 527 528 hio->hio_cmd = nv_get_uint8(nv, "cmd"); 529 if (hio->hio_cmd == 0) { 530 pjdlog_error("Header contains no 'cmd' field."); 531 hio->hio_error = EINVAL; 532 goto end; 533 } 534 if (hio->hio_cmd != HIO_KEEPALIVE) { 535 hio->hio_seq = nv_get_uint64(nv, "seq"); 536 if (hio->hio_seq == 0) { 537 pjdlog_error("Header contains no 'seq' field."); 538 hio->hio_error = EINVAL; 539 goto end; 540 } 541 } 542 switch (hio->hio_cmd) { 543 case HIO_FLUSH: 544 case HIO_KEEPALIVE: 545 break;
| 158init_environment(void) 159{ 160 struct hio *hio; 161 unsigned int ii; 162 163 /* 164 * Initialize lists, their locks and theirs condition variables. 165 */ 166 TAILQ_INIT(&hio_free_list); 167 mtx_init(&hio_free_list_lock); 168 cv_init(&hio_free_list_cond); 169 TAILQ_INIT(&hio_disk_list); 170 mtx_init(&hio_disk_list_lock); 171 cv_init(&hio_disk_list_cond); 172 TAILQ_INIT(&hio_send_list); 173 mtx_init(&hio_send_list_lock); 174 cv_init(&hio_send_list_cond); 175 176 /* 177 * Allocate requests pool and initialize requests. 178 */ 179 for (ii = 0; ii < HAST_HIO_MAX; ii++) { 180 hio = malloc(sizeof(*hio)); 181 if (hio == NULL) { 182 pjdlog_exitx(EX_TEMPFAIL, 183 "Unable to allocate memory (%zu bytes) for hio request.", 184 sizeof(*hio)); 185 } 186 hio->hio_data = malloc(MAXPHYS); 187 if (hio->hio_data == NULL) { 188 pjdlog_exitx(EX_TEMPFAIL, 189 "Unable to allocate memory (%zu bytes) for gctl_data.", 190 (size_t)MAXPHYS); 191 } 192 hio_clear(hio); 193 TAILQ_INSERT_HEAD(&hio_free_list, hio, hio_next); 194 } 195} 196 197static void 198init_local(struct hast_resource *res) 199{ 200 201 if (metadata_read(res, true) == -1) 202 exit(EX_NOINPUT); 203} 204 205static void 206init_remote(struct hast_resource *res, struct nv *nvin) 207{ 208 uint64_t resuid; 209 struct nv *nvout; 210 unsigned char *map; 211 size_t mapsize; 212 213#ifdef notyet 214 /* Setup direction. */ 215 if (proto_send(res->hr_remoteout, NULL, 0) == -1) 216 pjdlog_errno(LOG_WARNING, "Unable to set connection direction"); 217#endif 218 219 nvout = nv_alloc(); 220 nv_add_int64(nvout, (int64_t)res->hr_datasize, "datasize"); 221 nv_add_int32(nvout, (int32_t)res->hr_extentsize, "extentsize"); 222 resuid = nv_get_uint64(nvin, "resuid"); 223 res->hr_primary_localcnt = nv_get_uint64(nvin, "localcnt"); 224 res->hr_primary_remotecnt = nv_get_uint64(nvin, "remotecnt"); 225 nv_add_uint64(nvout, res->hr_secondary_localcnt, "localcnt"); 226 nv_add_uint64(nvout, res->hr_secondary_remotecnt, "remotecnt"); 227 mapsize = activemap_calc_ondisk_size(res->hr_local_mediasize - 228 METADATA_SIZE, res->hr_extentsize, res->hr_local_sectorsize); 229 map = malloc(mapsize); 230 if (map == NULL) { 231 pjdlog_exitx(EX_TEMPFAIL, 232 "Unable to allocate memory (%zu bytes) for activemap.", 233 mapsize); 234 } 235 /* 236 * When we work as primary and secondary is missing we will increase 237 * localcnt in our metadata. When secondary is connected and synced 238 * we make localcnt be equal to remotecnt, which means nodes are more 239 * or less in sync. 240 * Split-brain condition is when both nodes are not able to communicate 241 * and are both configured as primary nodes. In turn, they can both 242 * make incompatible changes to the data and we have to detect that. 243 * Under split-brain condition we will increase our localcnt on first 244 * write and remote node will increase its localcnt on first write. 245 * When we connect we can see that primary's localcnt is greater than 246 * our remotecnt (primary was modified while we weren't watching) and 247 * our localcnt is greater than primary's remotecnt (we were modified 248 * while primary wasn't watching). 249 * There are many possible combinations which are all gathered below. 250 * Don't pay too much attention to exact numbers, the more important 251 * is to compare them. We compare secondary's local with primary's 252 * remote and secondary's remote with primary's local. 253 * Note that every case where primary's localcnt is smaller than 254 * secondary's remotecnt and where secondary's localcnt is smaller than 255 * primary's remotecnt should be impossible in practise. We will perform 256 * full synchronization then. Those cases are marked with an asterisk. 257 * Regular synchronization means that only extents marked as dirty are 258 * synchronized (regular synchronization). 259 * 260 * SECONDARY METADATA PRIMARY METADATA 261 * local=3 remote=3 local=2 remote=2* ?! Full sync from secondary. 262 * local=3 remote=3 local=2 remote=3* ?! Full sync from primary. 263 * local=3 remote=3 local=2 remote=4* ?! Full sync from primary. 264 * local=3 remote=3 local=3 remote=2 Primary is out-of-date, 265 * regular sync from secondary. 266 * local=3 remote=3 local=3 remote=3 Regular sync just in case. 267 * local=3 remote=3 local=3 remote=4* ?! Full sync from primary. 268 * local=3 remote=3 local=4 remote=2 Split-brain condition. 269 * local=3 remote=3 local=4 remote=3 Secondary out-of-date, 270 * regular sync from primary. 271 * local=3 remote=3 local=4 remote=4* ?! Full sync from primary. 272 */ 273 if (res->hr_resuid == 0) { 274 /* 275 * Provider is used for the first time. If primary node done no 276 * writes yet as well (we will find "virgin" argument) then 277 * there is no need to synchronize anything. If primary node 278 * done any writes already we have to synchronize everything. 279 */ 280 PJDLOG_ASSERT(res->hr_secondary_localcnt == 0); 281 res->hr_resuid = resuid; 282 if (metadata_write(res) == -1) 283 exit(EX_NOINPUT); 284 if (nv_exists(nvin, "virgin")) { 285 free(map); 286 map = NULL; 287 mapsize = 0; 288 } else { 289 memset(map, 0xff, mapsize); 290 } 291 nv_add_int8(nvout, 1, "virgin"); 292 nv_add_uint8(nvout, HAST_SYNCSRC_PRIMARY, "syncsrc"); 293 } else if (res->hr_resuid != resuid) { 294 char errmsg[256]; 295 296 free(map); 297 (void)snprintf(errmsg, sizeof(errmsg), 298 "Resource unique ID mismatch (primary=%ju, secondary=%ju).", 299 (uintmax_t)resuid, (uintmax_t)res->hr_resuid); 300 pjdlog_error("%s", errmsg); 301 nv_add_string(nvout, errmsg, "errmsg"); 302 if (hast_proto_send(res, res->hr_remotein, nvout, 303 NULL, 0) == -1) { 304 pjdlog_exit(EX_TEMPFAIL, 305 "Unable to send response to %s", 306 res->hr_remoteaddr); 307 } 308 nv_free(nvout); 309 exit(EX_CONFIG); 310 } else if ( 311 /* Is primary out-of-date? */ 312 (res->hr_secondary_localcnt > res->hr_primary_remotecnt && 313 res->hr_secondary_remotecnt == res->hr_primary_localcnt) || 314 /* Are the nodes more or less in sync? */ 315 (res->hr_secondary_localcnt == res->hr_primary_remotecnt && 316 res->hr_secondary_remotecnt == res->hr_primary_localcnt) || 317 /* Is secondary out-of-date? */ 318 (res->hr_secondary_localcnt == res->hr_primary_remotecnt && 319 res->hr_secondary_remotecnt < res->hr_primary_localcnt)) { 320 /* 321 * Nodes are more or less in sync or one of the nodes is 322 * out-of-date. 323 * It doesn't matter at this point which one, we just have to 324 * send out local bitmap to the remote node. 325 */ 326 if (pread(res->hr_localfd, map, mapsize, METADATA_SIZE) != 327 (ssize_t)mapsize) { 328 pjdlog_exit(LOG_ERR, "Unable to read activemap"); 329 } 330 if (res->hr_secondary_localcnt > res->hr_primary_remotecnt && 331 res->hr_secondary_remotecnt == res->hr_primary_localcnt) { 332 /* Primary is out-of-date, sync from secondary. */ 333 nv_add_uint8(nvout, HAST_SYNCSRC_SECONDARY, "syncsrc"); 334 } else { 335 /* 336 * Secondary is out-of-date or counts match. 337 * Sync from primary. 338 */ 339 nv_add_uint8(nvout, HAST_SYNCSRC_PRIMARY, "syncsrc"); 340 } 341 } else if (res->hr_secondary_localcnt > res->hr_primary_remotecnt && 342 res->hr_primary_localcnt > res->hr_secondary_remotecnt) { 343 /* 344 * Not good, we have split-brain condition. 345 */ 346 free(map); 347 pjdlog_error("Split-brain detected, exiting."); 348 nv_add_string(nvout, "Split-brain condition!", "errmsg"); 349 if (hast_proto_send(res, res->hr_remotein, nvout, 350 NULL, 0) == -1) { 351 pjdlog_exit(EX_TEMPFAIL, 352 "Unable to send response to %s", 353 res->hr_remoteaddr); 354 } 355 nv_free(nvout); 356 /* Exit on split-brain. */ 357 event_send(res, EVENT_SPLITBRAIN); 358 exit(EX_CONFIG); 359 } else /* if (res->hr_secondary_localcnt < res->hr_primary_remotecnt || 360 res->hr_primary_localcnt < res->hr_secondary_remotecnt) */ { 361 /* 362 * This should never happen in practise, but we will perform 363 * full synchronization. 364 */ 365 PJDLOG_ASSERT(res->hr_secondary_localcnt < res->hr_primary_remotecnt || 366 res->hr_primary_localcnt < res->hr_secondary_remotecnt); 367 mapsize = activemap_calc_ondisk_size(res->hr_local_mediasize - 368 METADATA_SIZE, res->hr_extentsize, 369 res->hr_local_sectorsize); 370 memset(map, 0xff, mapsize); 371 if (res->hr_secondary_localcnt > res->hr_primary_remotecnt) { 372 /* In this one of five cases sync from secondary. */ 373 nv_add_uint8(nvout, HAST_SYNCSRC_SECONDARY, "syncsrc"); 374 } else { 375 /* For the rest four cases sync from primary. */ 376 nv_add_uint8(nvout, HAST_SYNCSRC_PRIMARY, "syncsrc"); 377 } 378 pjdlog_warning("This should never happen, asking for full synchronization (primary(local=%ju, remote=%ju), secondary(local=%ju, remote=%ju)).", 379 (uintmax_t)res->hr_primary_localcnt, 380 (uintmax_t)res->hr_primary_remotecnt, 381 (uintmax_t)res->hr_secondary_localcnt, 382 (uintmax_t)res->hr_secondary_remotecnt); 383 } 384 nv_add_uint32(nvout, (uint32_t)mapsize, "mapsize"); 385 if (hast_proto_send(res, res->hr_remotein, nvout, map, mapsize) == -1) { 386 pjdlog_exit(EX_TEMPFAIL, "Unable to send activemap to %s", 387 res->hr_remoteaddr); 388 } 389 if (map != NULL) 390 free(map); 391 nv_free(nvout); 392#ifdef notyet 393 /* Setup direction. */ 394 if (proto_recv(res->hr_remotein, NULL, 0) == -1) 395 pjdlog_errno(LOG_WARNING, "Unable to set connection direction"); 396#endif 397} 398 399void 400hastd_secondary(struct hast_resource *res, struct nv *nvin) 401{ 402 sigset_t mask; 403 pthread_t td; 404 pid_t pid; 405 int error, mode, debuglevel; 406 407 /* 408 * Create communication channel between parent and child. 409 */ 410 if (proto_client(NULL, "socketpair://", &res->hr_ctrl) == -1) { 411 KEEP_ERRNO((void)pidfile_remove(pfh)); 412 pjdlog_exit(EX_OSERR, 413 "Unable to create control sockets between parent and child"); 414 } 415 /* 416 * Create communication channel between child and parent. 417 */ 418 if (proto_client(NULL, "socketpair://", &res->hr_event) == -1) { 419 KEEP_ERRNO((void)pidfile_remove(pfh)); 420 pjdlog_exit(EX_OSERR, 421 "Unable to create event sockets between child and parent"); 422 } 423 424 pid = fork(); 425 if (pid == -1) { 426 KEEP_ERRNO((void)pidfile_remove(pfh)); 427 pjdlog_exit(EX_OSERR, "Unable to fork"); 428 } 429 430 if (pid > 0) { 431 /* This is parent. */ 432 proto_close(res->hr_remotein); 433 res->hr_remotein = NULL; 434 proto_close(res->hr_remoteout); 435 res->hr_remoteout = NULL; 436 /* Declare that we are receiver. */ 437 proto_recv(res->hr_event, NULL, 0); 438 /* Declare that we are sender. */ 439 proto_send(res->hr_ctrl, NULL, 0); 440 res->hr_workerpid = pid; 441 return; 442 } 443 444 gres = res; 445 mode = pjdlog_mode_get(); 446 debuglevel = pjdlog_debug_get(); 447 448 /* Declare that we are sender. */ 449 proto_send(res->hr_event, NULL, 0); 450 /* Declare that we are receiver. */ 451 proto_recv(res->hr_ctrl, NULL, 0); 452 descriptors_cleanup(res); 453 454 descriptors_assert(res, mode); 455 456 pjdlog_init(mode); 457 pjdlog_debug_set(debuglevel); 458 pjdlog_prefix_set("[%s] (%s) ", res->hr_name, role2str(res->hr_role)); 459 setproctitle("%s (%s)", res->hr_name, role2str(res->hr_role)); 460 461 PJDLOG_VERIFY(sigemptyset(&mask) == 0); 462 PJDLOG_VERIFY(sigprocmask(SIG_SETMASK, &mask, NULL) == 0); 463 464 /* Error in setting timeout is not critical, but why should it fail? */ 465 if (proto_timeout(res->hr_remotein, 2 * HAST_KEEPALIVE) == -1) 466 pjdlog_errno(LOG_WARNING, "Unable to set connection timeout"); 467 if (proto_timeout(res->hr_remoteout, res->hr_timeout) == -1) 468 pjdlog_errno(LOG_WARNING, "Unable to set connection timeout"); 469 470 init_local(res); 471 init_environment(); 472 473 if (drop_privs(res) != 0) 474 exit(EX_CONFIG); 475 pjdlog_info("Privileges successfully dropped."); 476 477 /* 478 * Create the control thread before sending any event to the parent, 479 * as we can deadlock when parent sends control request to worker, 480 * but worker has no control thread started yet, so parent waits. 481 * In the meantime worker sends an event to the parent, but parent 482 * is unable to handle the event, because it waits for control 483 * request response. 484 */ 485 error = pthread_create(&td, NULL, ctrl_thread, res); 486 PJDLOG_ASSERT(error == 0); 487 488 init_remote(res, nvin); 489 event_send(res, EVENT_CONNECT); 490 491 error = pthread_create(&td, NULL, recv_thread, res); 492 PJDLOG_ASSERT(error == 0); 493 error = pthread_create(&td, NULL, disk_thread, res); 494 PJDLOG_ASSERT(error == 0); 495 (void)send_thread(res); 496} 497 498static void 499reqlog(int loglevel, int debuglevel, int error, struct hio *hio, 500 const char *fmt, ...) 501{ 502 char msg[1024]; 503 va_list ap; 504 int len; 505 506 va_start(ap, fmt); 507 len = vsnprintf(msg, sizeof(msg), fmt, ap); 508 va_end(ap); 509 if ((size_t)len < sizeof(msg)) { 510 switch (hio->hio_cmd) { 511 case HIO_READ: 512 (void)snprintf(msg + len, sizeof(msg) - len, 513 "READ(%ju, %ju).", (uintmax_t)hio->hio_offset, 514 (uintmax_t)hio->hio_length); 515 break; 516 case HIO_DELETE: 517 (void)snprintf(msg + len, sizeof(msg) - len, 518 "DELETE(%ju, %ju).", (uintmax_t)hio->hio_offset, 519 (uintmax_t)hio->hio_length); 520 break; 521 case HIO_FLUSH: 522 (void)snprintf(msg + len, sizeof(msg) - len, "FLUSH."); 523 break; 524 case HIO_WRITE: 525 (void)snprintf(msg + len, sizeof(msg) - len, 526 "WRITE(%ju, %ju).", (uintmax_t)hio->hio_offset, 527 (uintmax_t)hio->hio_length); 528 break; 529 case HIO_KEEPALIVE: 530 (void)snprintf(msg + len, sizeof(msg) - len, "KEEPALIVE."); 531 break; 532 default: 533 (void)snprintf(msg + len, sizeof(msg) - len, 534 "UNKNOWN(%u).", (unsigned int)hio->hio_cmd); 535 break; 536 } 537 } 538 pjdlog_common(loglevel, debuglevel, error, "%s", msg); 539} 540 541static int 542requnpack(struct hast_resource *res, struct hio *hio, struct nv *nv) 543{ 544 545 hio->hio_cmd = nv_get_uint8(nv, "cmd"); 546 if (hio->hio_cmd == 0) { 547 pjdlog_error("Header contains no 'cmd' field."); 548 hio->hio_error = EINVAL; 549 goto end; 550 } 551 if (hio->hio_cmd != HIO_KEEPALIVE) { 552 hio->hio_seq = nv_get_uint64(nv, "seq"); 553 if (hio->hio_seq == 0) { 554 pjdlog_error("Header contains no 'seq' field."); 555 hio->hio_error = EINVAL; 556 goto end; 557 } 558 } 559 switch (hio->hio_cmd) { 560 case HIO_FLUSH: 561 case HIO_KEEPALIVE: 562 break;
|
678 } 679 nv_free(nv); 680 pjdlog_debug(2, "recv: (%p) Moving request to the disk queue.", 681 hio); 682 QUEUE_INSERT(disk, hio); 683 } 684 /* NOTREACHED */ 685 return (NULL); 686} 687 688/* 689 * Thread reads from or writes to local component and also handles DELETE and 690 * FLUSH requests. 691 */ 692static void * 693disk_thread(void *arg) 694{ 695 struct hast_resource *res = arg; 696 struct hio *hio; 697 ssize_t ret; 698 bool clear_activemap, logerror; 699 700 clear_activemap = true; 701 702 for (;;) { 703 pjdlog_debug(2, "disk: Taking request."); 704 QUEUE_TAKE(disk, hio); 705 while (clear_activemap) { 706 unsigned char *map; 707 size_t mapsize; 708 709 /* 710 * When first request is received, it means that primary 711 * already received our activemap, merged it and stored 712 * locally. We can now safely clear our activemap. 713 */ 714 mapsize = 715 activemap_calc_ondisk_size(res->hr_local_mediasize - 716 METADATA_SIZE, res->hr_extentsize, 717 res->hr_local_sectorsize); 718 map = calloc(1, mapsize); 719 if (map == NULL) { 720 pjdlog_warning("Unable to allocate memory to clear local activemap."); 721 break; 722 } 723 if (pwrite(res->hr_localfd, map, mapsize, 724 METADATA_SIZE) != (ssize_t)mapsize) { 725 pjdlog_errno(LOG_WARNING, 726 "Unable to store cleared activemap"); 727 free(map); 728 break; 729 } 730 free(map); 731 clear_activemap = false; 732 pjdlog_debug(1, "Local activemap cleared."); 733 break; 734 } 735 reqlog(LOG_DEBUG, 2, -1, hio, "disk: (%p) Got request: ", hio); 736 logerror = true; 737 /* Handle the actual request. */ 738 switch (hio->hio_cmd) { 739 case HIO_READ: 740 ret = pread(res->hr_localfd, hio->hio_data, 741 hio->hio_length, 742 hio->hio_offset + res->hr_localoff); 743 if (ret == -1) 744 hio->hio_error = errno; 745 else if (ret != (int64_t)hio->hio_length) 746 hio->hio_error = EIO; 747 else 748 hio->hio_error = 0; 749 break; 750 case HIO_WRITE: 751 ret = pwrite(res->hr_localfd, hio->hio_data, 752 hio->hio_length, 753 hio->hio_offset + res->hr_localoff); 754 if (ret == -1) 755 hio->hio_error = errno; 756 else if (ret != (int64_t)hio->hio_length) 757 hio->hio_error = EIO; 758 else 759 hio->hio_error = 0; 760 break; 761 case HIO_DELETE: 762 ret = g_delete(res->hr_localfd, 763 hio->hio_offset + res->hr_localoff, 764 hio->hio_length); 765 if (ret == -1) 766 hio->hio_error = errno; 767 else 768 hio->hio_error = 0; 769 break; 770 case HIO_FLUSH: 771 if (!res->hr_localflush) { 772 ret = -1; 773 hio->hio_error = EOPNOTSUPP; 774 logerror = false; 775 break; 776 } 777 ret = g_flush(res->hr_localfd); 778 if (ret == -1) { 779 if (errno == EOPNOTSUPP) 780 res->hr_localflush = false; 781 hio->hio_error = errno; 782 } else { 783 hio->hio_error = 0; 784 } 785 break; 786 default: 787 PJDLOG_ABORT("Unexpected command (cmd=%hhu).", 788 hio->hio_cmd); 789 } 790 if (logerror && hio->hio_error != 0) { 791 reqlog(LOG_ERR, 0, hio->hio_error, hio, 792 "Request failed: "); 793 } 794 pjdlog_debug(2, "disk: (%p) Moving request to the send queue.", 795 hio); 796 QUEUE_INSERT(send, hio); 797 } 798 /* NOTREACHED */ 799 return (NULL); 800} 801 802/* 803 * Thread sends requests back to primary node. 804 */ 805static void * 806send_thread(void *arg) 807{ 808 struct hast_resource *res = arg; 809 struct nv *nvout; 810 struct hio *hio; 811 void *data; 812 size_t length; 813 814 for (;;) { 815 pjdlog_debug(2, "send: Taking request."); 816 QUEUE_TAKE(send, hio); 817 reqlog(LOG_DEBUG, 2, -1, hio, "send: (%p) Got request: ", hio); 818 nvout = nv_alloc(); 819 /* Copy sequence number. */ 820 nv_add_uint64(nvout, hio->hio_seq, "seq");
| 718 } 719 nv_free(nv); 720 pjdlog_debug(2, "recv: (%p) Moving request to the disk queue.", 721 hio); 722 QUEUE_INSERT(disk, hio); 723 } 724 /* NOTREACHED */ 725 return (NULL); 726} 727 728/* 729 * Thread reads from or writes to local component and also handles DELETE and 730 * FLUSH requests. 731 */ 732static void * 733disk_thread(void *arg) 734{ 735 struct hast_resource *res = arg; 736 struct hio *hio; 737 ssize_t ret; 738 bool clear_activemap, logerror; 739 740 clear_activemap = true; 741 742 for (;;) { 743 pjdlog_debug(2, "disk: Taking request."); 744 QUEUE_TAKE(disk, hio); 745 while (clear_activemap) { 746 unsigned char *map; 747 size_t mapsize; 748 749 /* 750 * When first request is received, it means that primary 751 * already received our activemap, merged it and stored 752 * locally. We can now safely clear our activemap. 753 */ 754 mapsize = 755 activemap_calc_ondisk_size(res->hr_local_mediasize - 756 METADATA_SIZE, res->hr_extentsize, 757 res->hr_local_sectorsize); 758 map = calloc(1, mapsize); 759 if (map == NULL) { 760 pjdlog_warning("Unable to allocate memory to clear local activemap."); 761 break; 762 } 763 if (pwrite(res->hr_localfd, map, mapsize, 764 METADATA_SIZE) != (ssize_t)mapsize) { 765 pjdlog_errno(LOG_WARNING, 766 "Unable to store cleared activemap"); 767 free(map); 768 break; 769 } 770 free(map); 771 clear_activemap = false; 772 pjdlog_debug(1, "Local activemap cleared."); 773 break; 774 } 775 reqlog(LOG_DEBUG, 2, -1, hio, "disk: (%p) Got request: ", hio); 776 logerror = true; 777 /* Handle the actual request. */ 778 switch (hio->hio_cmd) { 779 case HIO_READ: 780 ret = pread(res->hr_localfd, hio->hio_data, 781 hio->hio_length, 782 hio->hio_offset + res->hr_localoff); 783 if (ret == -1) 784 hio->hio_error = errno; 785 else if (ret != (int64_t)hio->hio_length) 786 hio->hio_error = EIO; 787 else 788 hio->hio_error = 0; 789 break; 790 case HIO_WRITE: 791 ret = pwrite(res->hr_localfd, hio->hio_data, 792 hio->hio_length, 793 hio->hio_offset + res->hr_localoff); 794 if (ret == -1) 795 hio->hio_error = errno; 796 else if (ret != (int64_t)hio->hio_length) 797 hio->hio_error = EIO; 798 else 799 hio->hio_error = 0; 800 break; 801 case HIO_DELETE: 802 ret = g_delete(res->hr_localfd, 803 hio->hio_offset + res->hr_localoff, 804 hio->hio_length); 805 if (ret == -1) 806 hio->hio_error = errno; 807 else 808 hio->hio_error = 0; 809 break; 810 case HIO_FLUSH: 811 if (!res->hr_localflush) { 812 ret = -1; 813 hio->hio_error = EOPNOTSUPP; 814 logerror = false; 815 break; 816 } 817 ret = g_flush(res->hr_localfd); 818 if (ret == -1) { 819 if (errno == EOPNOTSUPP) 820 res->hr_localflush = false; 821 hio->hio_error = errno; 822 } else { 823 hio->hio_error = 0; 824 } 825 break; 826 default: 827 PJDLOG_ABORT("Unexpected command (cmd=%hhu).", 828 hio->hio_cmd); 829 } 830 if (logerror && hio->hio_error != 0) { 831 reqlog(LOG_ERR, 0, hio->hio_error, hio, 832 "Request failed: "); 833 } 834 pjdlog_debug(2, "disk: (%p) Moving request to the send queue.", 835 hio); 836 QUEUE_INSERT(send, hio); 837 } 838 /* NOTREACHED */ 839 return (NULL); 840} 841 842/* 843 * Thread sends requests back to primary node. 844 */ 845static void * 846send_thread(void *arg) 847{ 848 struct hast_resource *res = arg; 849 struct nv *nvout; 850 struct hio *hio; 851 void *data; 852 size_t length; 853 854 for (;;) { 855 pjdlog_debug(2, "send: Taking request."); 856 QUEUE_TAKE(send, hio); 857 reqlog(LOG_DEBUG, 2, -1, hio, "send: (%p) Got request: ", hio); 858 nvout = nv_alloc(); 859 /* Copy sequence number. */ 860 nv_add_uint64(nvout, hio->hio_seq, "seq");
|