32
33#include <sys/types.h>
34#include <sys/time.h>
35#include <sys/bio.h>
36#include <sys/disk.h>
37#include <sys/refcount.h>
38#include <sys/stat.h>
39
40#include <geom/gate/g_gate.h>
41
42#include <assert.h>
43#include <err.h>
44#include <errno.h>
45#include <fcntl.h>
46#include <libgeom.h>
47#include <pthread.h>
48#include <stdint.h>
49#include <stdio.h>
50#include <string.h>
51#include <sysexits.h>
52#include <unistd.h>
53
54#include <activemap.h>
55#include <nv.h>
56#include <rangelock.h>
57
58#include "control.h"
59#include "hast.h"
60#include "hast_proto.h"
61#include "hastd.h"
62#include "metadata.h"
63#include "proto.h"
64#include "pjdlog.h"
65#include "subr.h"
66#include "synch.h"
67
68struct hio {
69 /*
70 * Number of components we are still waiting for.
71 * When this field goes to 0, we can send the request back to the
72 * kernel. Each component has to decrease this counter by one
73 * even on failure.
74 */
75 unsigned int hio_countdown;
76 /*
77 * Each component has a place to store its own error.
78 * Once the request is handled by all components we can decide if the
79 * request overall is successful or not.
80 */
81 int *hio_errors;
82 /*
83 * Structure used to comunicate with GEOM Gate class.
84 */
85 struct g_gate_ctl_io hio_ggio;
86 TAILQ_ENTRY(hio) *hio_next;
87};
88#define hio_free_next hio_next[0]
89#define hio_done_next hio_next[0]
90
91/*
92 * Free list holds unused structures. When free list is empty, we have to wait
93 * until some in-progress requests are freed.
94 */
95static TAILQ_HEAD(, hio) hio_free_list;
96static pthread_mutex_t hio_free_list_lock;
97static pthread_cond_t hio_free_list_cond;
98/*
99 * There is one send list for every component. One requests is placed on all
100 * send lists - each component gets the same request, but each component is
101 * responsible for managing his own send list.
102 */
103static TAILQ_HEAD(, hio) *hio_send_list;
104static pthread_mutex_t *hio_send_list_lock;
105static pthread_cond_t *hio_send_list_cond;
106/*
107 * There is one recv list for every component, although local components don't
108 * use recv lists as local requests are done synchronously.
109 */
110static TAILQ_HEAD(, hio) *hio_recv_list;
111static pthread_mutex_t *hio_recv_list_lock;
112static pthread_cond_t *hio_recv_list_cond;
113/*
114 * Request is placed on done list by the slowest component (the one that
115 * decreased hio_countdown from 1 to 0).
116 */
117static TAILQ_HEAD(, hio) hio_done_list;
118static pthread_mutex_t hio_done_list_lock;
119static pthread_cond_t hio_done_list_cond;
120/*
121 * Structure below are for interaction with sync thread.
122 */
123static bool sync_inprogress;
124static pthread_mutex_t sync_lock;
125static pthread_cond_t sync_cond;
126/*
127 * The lock below allows to synchornize access to remote connections.
128 */
129static pthread_rwlock_t *hio_remote_lock;
130static pthread_mutex_t hio_guard_lock;
131static pthread_cond_t hio_guard_cond;
132
133/*
134 * Lock to synchronize metadata updates. Also synchronize access to
135 * hr_primary_localcnt and hr_primary_remotecnt fields.
136 */
137static pthread_mutex_t metadata_lock;
138
139/*
140 * Maximum number of outstanding I/O requests.
141 */
142#define HAST_HIO_MAX 256
143/*
144 * Number of components. At this point there are only two components: local
145 * and remote, but in the future it might be possible to use multiple local
146 * and remote components.
147 */
148#define HAST_NCOMPONENTS 2
149/*
150 * Number of seconds to sleep before next reconnect try.
151 */
152#define RECONNECT_SLEEP 5
153
154#define ISCONNECTED(res, no) \
155 ((res)->hr_remotein != NULL && (res)->hr_remoteout != NULL)
156
157#define QUEUE_INSERT1(hio, name, ncomp) do { \
158 bool _wakeup; \
159 \
160 mtx_lock(&hio_##name##_list_lock[(ncomp)]); \
161 _wakeup = TAILQ_EMPTY(&hio_##name##_list[(ncomp)]); \
162 TAILQ_INSERT_TAIL(&hio_##name##_list[(ncomp)], (hio), \
163 hio_next[(ncomp)]); \
164 mtx_unlock(&hio_##name##_list_lock[ncomp]); \
165 if (_wakeup) \
166 cv_signal(&hio_##name##_list_cond[(ncomp)]); \
167} while (0)
168#define QUEUE_INSERT2(hio, name) do { \
169 bool _wakeup; \
170 \
171 mtx_lock(&hio_##name##_list_lock); \
172 _wakeup = TAILQ_EMPTY(&hio_##name##_list); \
173 TAILQ_INSERT_TAIL(&hio_##name##_list, (hio), hio_##name##_next);\
174 mtx_unlock(&hio_##name##_list_lock); \
175 if (_wakeup) \
176 cv_signal(&hio_##name##_list_cond); \
177} while (0)
178#define QUEUE_TAKE1(hio, name, ncomp) do { \
179 mtx_lock(&hio_##name##_list_lock[(ncomp)]); \
180 while (((hio) = TAILQ_FIRST(&hio_##name##_list[(ncomp)])) == NULL) { \
181 cv_wait(&hio_##name##_list_cond[(ncomp)], \
182 &hio_##name##_list_lock[(ncomp)]); \
183 } \
184 TAILQ_REMOVE(&hio_##name##_list[(ncomp)], (hio), \
185 hio_next[(ncomp)]); \
186 mtx_unlock(&hio_##name##_list_lock[(ncomp)]); \
187} while (0)
188#define QUEUE_TAKE2(hio, name) do { \
189 mtx_lock(&hio_##name##_list_lock); \
190 while (((hio) = TAILQ_FIRST(&hio_##name##_list)) == NULL) { \
191 cv_wait(&hio_##name##_list_cond, \
192 &hio_##name##_list_lock); \
193 } \
194 TAILQ_REMOVE(&hio_##name##_list, (hio), hio_##name##_next); \
195 mtx_unlock(&hio_##name##_list_lock); \
196} while (0)
197
198#define SYNCREQ(hio) do { (hio)->hio_ggio.gctl_unit = -1; } while (0)
199#define ISSYNCREQ(hio) ((hio)->hio_ggio.gctl_unit == -1)
200#define SYNCREQDONE(hio) do { (hio)->hio_ggio.gctl_unit = -2; } while (0)
201#define ISSYNCREQDONE(hio) ((hio)->hio_ggio.gctl_unit == -2)
202
203static struct hast_resource *gres;
204
205static pthread_mutex_t range_lock;
206static struct rangelocks *range_regular;
207static bool range_regular_wait;
208static pthread_cond_t range_regular_cond;
209static struct rangelocks *range_sync;
210static bool range_sync_wait;
211static pthread_cond_t range_sync_cond;
212
213static void *ggate_recv_thread(void *arg);
214static void *local_send_thread(void *arg);
215static void *remote_send_thread(void *arg);
216static void *remote_recv_thread(void *arg);
217static void *ggate_send_thread(void *arg);
218static void *sync_thread(void *arg);
219static void *guard_thread(void *arg);
220
221static void sighandler(int sig);
222
223static void
224cleanup(struct hast_resource *res)
225{
226 int rerrno;
227
228 /* Remember errno. */
229 rerrno = errno;
230
231 /*
232 * Close descriptor to /dev/hast/<name>
233 * to work-around race in the kernel.
234 */
235 close(res->hr_localfd);
236
237 /* Destroy ggate provider if we created one. */
238 if (res->hr_ggateunit >= 0) {
239 struct g_gate_ctl_destroy ggiod;
240
241 ggiod.gctl_version = G_GATE_VERSION;
242 ggiod.gctl_unit = res->hr_ggateunit;
243 ggiod.gctl_force = 1;
244 if (ioctl(res->hr_ggatefd, G_GATE_CMD_DESTROY, &ggiod) < 0) {
245 pjdlog_warning("Unable to destroy hast/%s device",
246 res->hr_provname);
247 }
248 res->hr_ggateunit = -1;
249 }
250
251 /* Restore errno. */
252 errno = rerrno;
253}
254
255static void
256primary_exit(int exitcode, const char *fmt, ...)
257{
258 va_list ap;
259
260 assert(exitcode != EX_OK);
261 va_start(ap, fmt);
262 pjdlogv_errno(LOG_ERR, fmt, ap);
263 va_end(ap);
264 cleanup(gres);
265 exit(exitcode);
266}
267
268static void
269primary_exitx(int exitcode, const char *fmt, ...)
270{
271 va_list ap;
272
273 va_start(ap, fmt);
274 pjdlogv(exitcode == EX_OK ? LOG_INFO : LOG_ERR, fmt, ap);
275 va_end(ap);
276 cleanup(gres);
277 exit(exitcode);
278}
279
280static int
281hast_activemap_flush(struct hast_resource *res)
282{
283 const unsigned char *buf;
284 size_t size;
285
286 buf = activemap_bitmap(res->hr_amp, &size);
287 assert(buf != NULL);
288 assert((size % res->hr_local_sectorsize) == 0);
289 if (pwrite(res->hr_localfd, buf, size, METADATA_SIZE) !=
290 (ssize_t)size) {
291 KEEP_ERRNO(pjdlog_errno(LOG_ERR,
292 "Unable to flush activemap to disk"));
293 return (-1);
294 }
295 return (0);
296}
297
298static void
299init_environment(struct hast_resource *res __unused)
300{
301 struct hio *hio;
302 unsigned int ii, ncomps;
303
304 /*
305 * In the future it might be per-resource value.
306 */
307 ncomps = HAST_NCOMPONENTS;
308
309 /*
310 * Allocate memory needed by lists.
311 */
312 hio_send_list = malloc(sizeof(hio_send_list[0]) * ncomps);
313 if (hio_send_list == NULL) {
314 primary_exitx(EX_TEMPFAIL,
315 "Unable to allocate %zu bytes of memory for send lists.",
316 sizeof(hio_send_list[0]) * ncomps);
317 }
318 hio_send_list_lock = malloc(sizeof(hio_send_list_lock[0]) * ncomps);
319 if (hio_send_list_lock == NULL) {
320 primary_exitx(EX_TEMPFAIL,
321 "Unable to allocate %zu bytes of memory for send list locks.",
322 sizeof(hio_send_list_lock[0]) * ncomps);
323 }
324 hio_send_list_cond = malloc(sizeof(hio_send_list_cond[0]) * ncomps);
325 if (hio_send_list_cond == NULL) {
326 primary_exitx(EX_TEMPFAIL,
327 "Unable to allocate %zu bytes of memory for send list condition variables.",
328 sizeof(hio_send_list_cond[0]) * ncomps);
329 }
330 hio_recv_list = malloc(sizeof(hio_recv_list[0]) * ncomps);
331 if (hio_recv_list == NULL) {
332 primary_exitx(EX_TEMPFAIL,
333 "Unable to allocate %zu bytes of memory for recv lists.",
334 sizeof(hio_recv_list[0]) * ncomps);
335 }
336 hio_recv_list_lock = malloc(sizeof(hio_recv_list_lock[0]) * ncomps);
337 if (hio_recv_list_lock == NULL) {
338 primary_exitx(EX_TEMPFAIL,
339 "Unable to allocate %zu bytes of memory for recv list locks.",
340 sizeof(hio_recv_list_lock[0]) * ncomps);
341 }
342 hio_recv_list_cond = malloc(sizeof(hio_recv_list_cond[0]) * ncomps);
343 if (hio_recv_list_cond == NULL) {
344 primary_exitx(EX_TEMPFAIL,
345 "Unable to allocate %zu bytes of memory for recv list condition variables.",
346 sizeof(hio_recv_list_cond[0]) * ncomps);
347 }
348 hio_remote_lock = malloc(sizeof(hio_remote_lock[0]) * ncomps);
349 if (hio_remote_lock == NULL) {
350 primary_exitx(EX_TEMPFAIL,
351 "Unable to allocate %zu bytes of memory for remote connections locks.",
352 sizeof(hio_remote_lock[0]) * ncomps);
353 }
354
355 /*
356 * Initialize lists, their locks and theirs condition variables.
357 */
358 TAILQ_INIT(&hio_free_list);
359 mtx_init(&hio_free_list_lock);
360 cv_init(&hio_free_list_cond);
361 for (ii = 0; ii < HAST_NCOMPONENTS; ii++) {
362 TAILQ_INIT(&hio_send_list[ii]);
363 mtx_init(&hio_send_list_lock[ii]);
364 cv_init(&hio_send_list_cond[ii]);
365 TAILQ_INIT(&hio_recv_list[ii]);
366 mtx_init(&hio_recv_list_lock[ii]);
367 cv_init(&hio_recv_list_cond[ii]);
368 rw_init(&hio_remote_lock[ii]);
369 }
370 TAILQ_INIT(&hio_done_list);
371 mtx_init(&hio_done_list_lock);
372 cv_init(&hio_done_list_cond);
373 mtx_init(&hio_guard_lock);
374 cv_init(&hio_guard_cond);
375 mtx_init(&metadata_lock);
376
377 /*
378 * Allocate requests pool and initialize requests.
379 */
380 for (ii = 0; ii < HAST_HIO_MAX; ii++) {
381 hio = malloc(sizeof(*hio));
382 if (hio == NULL) {
383 primary_exitx(EX_TEMPFAIL,
384 "Unable to allocate %zu bytes of memory for hio request.",
385 sizeof(*hio));
386 }
387 hio->hio_countdown = 0;
388 hio->hio_errors = malloc(sizeof(hio->hio_errors[0]) * ncomps);
389 if (hio->hio_errors == NULL) {
390 primary_exitx(EX_TEMPFAIL,
391 "Unable allocate %zu bytes of memory for hio errors.",
392 sizeof(hio->hio_errors[0]) * ncomps);
393 }
394 hio->hio_next = malloc(sizeof(hio->hio_next[0]) * ncomps);
395 if (hio->hio_next == NULL) {
396 primary_exitx(EX_TEMPFAIL,
397 "Unable allocate %zu bytes of memory for hio_next field.",
398 sizeof(hio->hio_next[0]) * ncomps);
399 }
400 hio->hio_ggio.gctl_version = G_GATE_VERSION;
401 hio->hio_ggio.gctl_data = malloc(MAXPHYS);
402 if (hio->hio_ggio.gctl_data == NULL) {
403 primary_exitx(EX_TEMPFAIL,
404 "Unable to allocate %zu bytes of memory for gctl_data.",
405 MAXPHYS);
406 }
407 hio->hio_ggio.gctl_length = MAXPHYS;
408 hio->hio_ggio.gctl_error = 0;
409 TAILQ_INSERT_HEAD(&hio_free_list, hio, hio_free_next);
410 }
411
412 /*
413 * Turn on signals handling.
414 */
415 signal(SIGINT, sighandler);
416 signal(SIGTERM, sighandler);
417}
418
419static void
420init_local(struct hast_resource *res)
421{
422 unsigned char *buf;
423 size_t mapsize;
424
425 if (metadata_read(res, true) < 0)
426 exit(EX_NOINPUT);
427 mtx_init(&res->hr_amp_lock);
428 if (activemap_init(&res->hr_amp, res->hr_datasize, res->hr_extentsize,
429 res->hr_local_sectorsize, res->hr_keepdirty) < 0) {
430 primary_exit(EX_TEMPFAIL, "Unable to create activemap");
431 }
432 mtx_init(&range_lock);
433 cv_init(&range_regular_cond);
434 if (rangelock_init(&range_regular) < 0)
435 primary_exit(EX_TEMPFAIL, "Unable to create regular range lock");
436 cv_init(&range_sync_cond);
437 if (rangelock_init(&range_sync) < 0)
438 primary_exit(EX_TEMPFAIL, "Unable to create sync range lock");
439 mapsize = activemap_ondisk_size(res->hr_amp);
440 buf = calloc(1, mapsize);
441 if (buf == NULL) {
442 primary_exitx(EX_TEMPFAIL,
443 "Unable to allocate buffer for activemap.");
444 }
445 if (pread(res->hr_localfd, buf, mapsize, METADATA_SIZE) !=
446 (ssize_t)mapsize) {
447 primary_exit(EX_NOINPUT, "Unable to read activemap");
448 }
449 activemap_copyin(res->hr_amp, buf, mapsize);
450 if (res->hr_resuid != 0)
451 return;
452 /*
453 * We're using provider for the first time, so we have to generate
454 * resource unique identifier and initialize local and remote counts.
455 */
456 arc4random_buf(&res->hr_resuid, sizeof(res->hr_resuid));
457 res->hr_primary_localcnt = 1;
458 res->hr_primary_remotecnt = 0;
459 if (metadata_write(res) < 0)
460 exit(EX_NOINPUT);
461}
462
463static bool
464init_remote(struct hast_resource *res, struct proto_conn **inp,
465 struct proto_conn **outp)
466{
467 struct proto_conn *in, *out;
468 struct nv *nvout, *nvin;
469 const unsigned char *token;
470 unsigned char *map;
471 const char *errmsg;
472 int32_t extentsize;
473 int64_t datasize;
474 uint32_t mapsize;
475 size_t size;
476
477 assert((inp == NULL && outp == NULL) || (inp != NULL && outp != NULL));
478
479 in = out = NULL;
480
481 /* Prepare outgoing connection with remote node. */
482 if (proto_client(res->hr_remoteaddr, &out) < 0) {
483 primary_exit(EX_TEMPFAIL, "Unable to create connection to %s",
484 res->hr_remoteaddr);
485 }
486 /* Try to connect, but accept failure. */
487 if (proto_connect(out) < 0) {
488 pjdlog_errno(LOG_WARNING, "Unable to connect to %s",
489 res->hr_remoteaddr);
490 goto close;
491 }
| 32
33#include <sys/types.h>
34#include <sys/time.h>
35#include <sys/bio.h>
36#include <sys/disk.h>
37#include <sys/refcount.h>
38#include <sys/stat.h>
39
40#include <geom/gate/g_gate.h>
41
42#include <assert.h>
43#include <err.h>
44#include <errno.h>
45#include <fcntl.h>
46#include <libgeom.h>
47#include <pthread.h>
48#include <stdint.h>
49#include <stdio.h>
50#include <string.h>
51#include <sysexits.h>
52#include <unistd.h>
53
54#include <activemap.h>
55#include <nv.h>
56#include <rangelock.h>
57
58#include "control.h"
59#include "hast.h"
60#include "hast_proto.h"
61#include "hastd.h"
62#include "metadata.h"
63#include "proto.h"
64#include "pjdlog.h"
65#include "subr.h"
66#include "synch.h"
67
68struct hio {
69 /*
70 * Number of components we are still waiting for.
71 * When this field goes to 0, we can send the request back to the
72 * kernel. Each component has to decrease this counter by one
73 * even on failure.
74 */
75 unsigned int hio_countdown;
76 /*
77 * Each component has a place to store its own error.
78 * Once the request is handled by all components we can decide if the
79 * request overall is successful or not.
80 */
81 int *hio_errors;
82 /*
83 * Structure used to comunicate with GEOM Gate class.
84 */
85 struct g_gate_ctl_io hio_ggio;
86 TAILQ_ENTRY(hio) *hio_next;
87};
88#define hio_free_next hio_next[0]
89#define hio_done_next hio_next[0]
90
91/*
92 * Free list holds unused structures. When free list is empty, we have to wait
93 * until some in-progress requests are freed.
94 */
95static TAILQ_HEAD(, hio) hio_free_list;
96static pthread_mutex_t hio_free_list_lock;
97static pthread_cond_t hio_free_list_cond;
98/*
99 * There is one send list for every component. One requests is placed on all
100 * send lists - each component gets the same request, but each component is
101 * responsible for managing his own send list.
102 */
103static TAILQ_HEAD(, hio) *hio_send_list;
104static pthread_mutex_t *hio_send_list_lock;
105static pthread_cond_t *hio_send_list_cond;
106/*
107 * There is one recv list for every component, although local components don't
108 * use recv lists as local requests are done synchronously.
109 */
110static TAILQ_HEAD(, hio) *hio_recv_list;
111static pthread_mutex_t *hio_recv_list_lock;
112static pthread_cond_t *hio_recv_list_cond;
113/*
114 * Request is placed on done list by the slowest component (the one that
115 * decreased hio_countdown from 1 to 0).
116 */
117static TAILQ_HEAD(, hio) hio_done_list;
118static pthread_mutex_t hio_done_list_lock;
119static pthread_cond_t hio_done_list_cond;
120/*
121 * Structure below are for interaction with sync thread.
122 */
123static bool sync_inprogress;
124static pthread_mutex_t sync_lock;
125static pthread_cond_t sync_cond;
126/*
127 * The lock below allows to synchornize access to remote connections.
128 */
129static pthread_rwlock_t *hio_remote_lock;
130static pthread_mutex_t hio_guard_lock;
131static pthread_cond_t hio_guard_cond;
132
133/*
134 * Lock to synchronize metadata updates. Also synchronize access to
135 * hr_primary_localcnt and hr_primary_remotecnt fields.
136 */
137static pthread_mutex_t metadata_lock;
138
139/*
140 * Maximum number of outstanding I/O requests.
141 */
142#define HAST_HIO_MAX 256
143/*
144 * Number of components. At this point there are only two components: local
145 * and remote, but in the future it might be possible to use multiple local
146 * and remote components.
147 */
148#define HAST_NCOMPONENTS 2
149/*
150 * Number of seconds to sleep before next reconnect try.
151 */
152#define RECONNECT_SLEEP 5
153
154#define ISCONNECTED(res, no) \
155 ((res)->hr_remotein != NULL && (res)->hr_remoteout != NULL)
156
157#define QUEUE_INSERT1(hio, name, ncomp) do { \
158 bool _wakeup; \
159 \
160 mtx_lock(&hio_##name##_list_lock[(ncomp)]); \
161 _wakeup = TAILQ_EMPTY(&hio_##name##_list[(ncomp)]); \
162 TAILQ_INSERT_TAIL(&hio_##name##_list[(ncomp)], (hio), \
163 hio_next[(ncomp)]); \
164 mtx_unlock(&hio_##name##_list_lock[ncomp]); \
165 if (_wakeup) \
166 cv_signal(&hio_##name##_list_cond[(ncomp)]); \
167} while (0)
168#define QUEUE_INSERT2(hio, name) do { \
169 bool _wakeup; \
170 \
171 mtx_lock(&hio_##name##_list_lock); \
172 _wakeup = TAILQ_EMPTY(&hio_##name##_list); \
173 TAILQ_INSERT_TAIL(&hio_##name##_list, (hio), hio_##name##_next);\
174 mtx_unlock(&hio_##name##_list_lock); \
175 if (_wakeup) \
176 cv_signal(&hio_##name##_list_cond); \
177} while (0)
178#define QUEUE_TAKE1(hio, name, ncomp) do { \
179 mtx_lock(&hio_##name##_list_lock[(ncomp)]); \
180 while (((hio) = TAILQ_FIRST(&hio_##name##_list[(ncomp)])) == NULL) { \
181 cv_wait(&hio_##name##_list_cond[(ncomp)], \
182 &hio_##name##_list_lock[(ncomp)]); \
183 } \
184 TAILQ_REMOVE(&hio_##name##_list[(ncomp)], (hio), \
185 hio_next[(ncomp)]); \
186 mtx_unlock(&hio_##name##_list_lock[(ncomp)]); \
187} while (0)
188#define QUEUE_TAKE2(hio, name) do { \
189 mtx_lock(&hio_##name##_list_lock); \
190 while (((hio) = TAILQ_FIRST(&hio_##name##_list)) == NULL) { \
191 cv_wait(&hio_##name##_list_cond, \
192 &hio_##name##_list_lock); \
193 } \
194 TAILQ_REMOVE(&hio_##name##_list, (hio), hio_##name##_next); \
195 mtx_unlock(&hio_##name##_list_lock); \
196} while (0)
197
198#define SYNCREQ(hio) do { (hio)->hio_ggio.gctl_unit = -1; } while (0)
199#define ISSYNCREQ(hio) ((hio)->hio_ggio.gctl_unit == -1)
200#define SYNCREQDONE(hio) do { (hio)->hio_ggio.gctl_unit = -2; } while (0)
201#define ISSYNCREQDONE(hio) ((hio)->hio_ggio.gctl_unit == -2)
202
203static struct hast_resource *gres;
204
205static pthread_mutex_t range_lock;
206static struct rangelocks *range_regular;
207static bool range_regular_wait;
208static pthread_cond_t range_regular_cond;
209static struct rangelocks *range_sync;
210static bool range_sync_wait;
211static pthread_cond_t range_sync_cond;
212
213static void *ggate_recv_thread(void *arg);
214static void *local_send_thread(void *arg);
215static void *remote_send_thread(void *arg);
216static void *remote_recv_thread(void *arg);
217static void *ggate_send_thread(void *arg);
218static void *sync_thread(void *arg);
219static void *guard_thread(void *arg);
220
221static void sighandler(int sig);
222
223static void
224cleanup(struct hast_resource *res)
225{
226 int rerrno;
227
228 /* Remember errno. */
229 rerrno = errno;
230
231 /*
232 * Close descriptor to /dev/hast/<name>
233 * to work-around race in the kernel.
234 */
235 close(res->hr_localfd);
236
237 /* Destroy ggate provider if we created one. */
238 if (res->hr_ggateunit >= 0) {
239 struct g_gate_ctl_destroy ggiod;
240
241 ggiod.gctl_version = G_GATE_VERSION;
242 ggiod.gctl_unit = res->hr_ggateunit;
243 ggiod.gctl_force = 1;
244 if (ioctl(res->hr_ggatefd, G_GATE_CMD_DESTROY, &ggiod) < 0) {
245 pjdlog_warning("Unable to destroy hast/%s device",
246 res->hr_provname);
247 }
248 res->hr_ggateunit = -1;
249 }
250
251 /* Restore errno. */
252 errno = rerrno;
253}
254
255static void
256primary_exit(int exitcode, const char *fmt, ...)
257{
258 va_list ap;
259
260 assert(exitcode != EX_OK);
261 va_start(ap, fmt);
262 pjdlogv_errno(LOG_ERR, fmt, ap);
263 va_end(ap);
264 cleanup(gres);
265 exit(exitcode);
266}
267
268static void
269primary_exitx(int exitcode, const char *fmt, ...)
270{
271 va_list ap;
272
273 va_start(ap, fmt);
274 pjdlogv(exitcode == EX_OK ? LOG_INFO : LOG_ERR, fmt, ap);
275 va_end(ap);
276 cleanup(gres);
277 exit(exitcode);
278}
279
280static int
281hast_activemap_flush(struct hast_resource *res)
282{
283 const unsigned char *buf;
284 size_t size;
285
286 buf = activemap_bitmap(res->hr_amp, &size);
287 assert(buf != NULL);
288 assert((size % res->hr_local_sectorsize) == 0);
289 if (pwrite(res->hr_localfd, buf, size, METADATA_SIZE) !=
290 (ssize_t)size) {
291 KEEP_ERRNO(pjdlog_errno(LOG_ERR,
292 "Unable to flush activemap to disk"));
293 return (-1);
294 }
295 return (0);
296}
297
298static void
299init_environment(struct hast_resource *res __unused)
300{
301 struct hio *hio;
302 unsigned int ii, ncomps;
303
304 /*
305 * In the future it might be per-resource value.
306 */
307 ncomps = HAST_NCOMPONENTS;
308
309 /*
310 * Allocate memory needed by lists.
311 */
312 hio_send_list = malloc(sizeof(hio_send_list[0]) * ncomps);
313 if (hio_send_list == NULL) {
314 primary_exitx(EX_TEMPFAIL,
315 "Unable to allocate %zu bytes of memory for send lists.",
316 sizeof(hio_send_list[0]) * ncomps);
317 }
318 hio_send_list_lock = malloc(sizeof(hio_send_list_lock[0]) * ncomps);
319 if (hio_send_list_lock == NULL) {
320 primary_exitx(EX_TEMPFAIL,
321 "Unable to allocate %zu bytes of memory for send list locks.",
322 sizeof(hio_send_list_lock[0]) * ncomps);
323 }
324 hio_send_list_cond = malloc(sizeof(hio_send_list_cond[0]) * ncomps);
325 if (hio_send_list_cond == NULL) {
326 primary_exitx(EX_TEMPFAIL,
327 "Unable to allocate %zu bytes of memory for send list condition variables.",
328 sizeof(hio_send_list_cond[0]) * ncomps);
329 }
330 hio_recv_list = malloc(sizeof(hio_recv_list[0]) * ncomps);
331 if (hio_recv_list == NULL) {
332 primary_exitx(EX_TEMPFAIL,
333 "Unable to allocate %zu bytes of memory for recv lists.",
334 sizeof(hio_recv_list[0]) * ncomps);
335 }
336 hio_recv_list_lock = malloc(sizeof(hio_recv_list_lock[0]) * ncomps);
337 if (hio_recv_list_lock == NULL) {
338 primary_exitx(EX_TEMPFAIL,
339 "Unable to allocate %zu bytes of memory for recv list locks.",
340 sizeof(hio_recv_list_lock[0]) * ncomps);
341 }
342 hio_recv_list_cond = malloc(sizeof(hio_recv_list_cond[0]) * ncomps);
343 if (hio_recv_list_cond == NULL) {
344 primary_exitx(EX_TEMPFAIL,
345 "Unable to allocate %zu bytes of memory for recv list condition variables.",
346 sizeof(hio_recv_list_cond[0]) * ncomps);
347 }
348 hio_remote_lock = malloc(sizeof(hio_remote_lock[0]) * ncomps);
349 if (hio_remote_lock == NULL) {
350 primary_exitx(EX_TEMPFAIL,
351 "Unable to allocate %zu bytes of memory for remote connections locks.",
352 sizeof(hio_remote_lock[0]) * ncomps);
353 }
354
355 /*
356 * Initialize lists, their locks and theirs condition variables.
357 */
358 TAILQ_INIT(&hio_free_list);
359 mtx_init(&hio_free_list_lock);
360 cv_init(&hio_free_list_cond);
361 for (ii = 0; ii < HAST_NCOMPONENTS; ii++) {
362 TAILQ_INIT(&hio_send_list[ii]);
363 mtx_init(&hio_send_list_lock[ii]);
364 cv_init(&hio_send_list_cond[ii]);
365 TAILQ_INIT(&hio_recv_list[ii]);
366 mtx_init(&hio_recv_list_lock[ii]);
367 cv_init(&hio_recv_list_cond[ii]);
368 rw_init(&hio_remote_lock[ii]);
369 }
370 TAILQ_INIT(&hio_done_list);
371 mtx_init(&hio_done_list_lock);
372 cv_init(&hio_done_list_cond);
373 mtx_init(&hio_guard_lock);
374 cv_init(&hio_guard_cond);
375 mtx_init(&metadata_lock);
376
377 /*
378 * Allocate requests pool and initialize requests.
379 */
380 for (ii = 0; ii < HAST_HIO_MAX; ii++) {
381 hio = malloc(sizeof(*hio));
382 if (hio == NULL) {
383 primary_exitx(EX_TEMPFAIL,
384 "Unable to allocate %zu bytes of memory for hio request.",
385 sizeof(*hio));
386 }
387 hio->hio_countdown = 0;
388 hio->hio_errors = malloc(sizeof(hio->hio_errors[0]) * ncomps);
389 if (hio->hio_errors == NULL) {
390 primary_exitx(EX_TEMPFAIL,
391 "Unable allocate %zu bytes of memory for hio errors.",
392 sizeof(hio->hio_errors[0]) * ncomps);
393 }
394 hio->hio_next = malloc(sizeof(hio->hio_next[0]) * ncomps);
395 if (hio->hio_next == NULL) {
396 primary_exitx(EX_TEMPFAIL,
397 "Unable allocate %zu bytes of memory for hio_next field.",
398 sizeof(hio->hio_next[0]) * ncomps);
399 }
400 hio->hio_ggio.gctl_version = G_GATE_VERSION;
401 hio->hio_ggio.gctl_data = malloc(MAXPHYS);
402 if (hio->hio_ggio.gctl_data == NULL) {
403 primary_exitx(EX_TEMPFAIL,
404 "Unable to allocate %zu bytes of memory for gctl_data.",
405 MAXPHYS);
406 }
407 hio->hio_ggio.gctl_length = MAXPHYS;
408 hio->hio_ggio.gctl_error = 0;
409 TAILQ_INSERT_HEAD(&hio_free_list, hio, hio_free_next);
410 }
411
412 /*
413 * Turn on signals handling.
414 */
415 signal(SIGINT, sighandler);
416 signal(SIGTERM, sighandler);
417}
418
419static void
420init_local(struct hast_resource *res)
421{
422 unsigned char *buf;
423 size_t mapsize;
424
425 if (metadata_read(res, true) < 0)
426 exit(EX_NOINPUT);
427 mtx_init(&res->hr_amp_lock);
428 if (activemap_init(&res->hr_amp, res->hr_datasize, res->hr_extentsize,
429 res->hr_local_sectorsize, res->hr_keepdirty) < 0) {
430 primary_exit(EX_TEMPFAIL, "Unable to create activemap");
431 }
432 mtx_init(&range_lock);
433 cv_init(&range_regular_cond);
434 if (rangelock_init(&range_regular) < 0)
435 primary_exit(EX_TEMPFAIL, "Unable to create regular range lock");
436 cv_init(&range_sync_cond);
437 if (rangelock_init(&range_sync) < 0)
438 primary_exit(EX_TEMPFAIL, "Unable to create sync range lock");
439 mapsize = activemap_ondisk_size(res->hr_amp);
440 buf = calloc(1, mapsize);
441 if (buf == NULL) {
442 primary_exitx(EX_TEMPFAIL,
443 "Unable to allocate buffer for activemap.");
444 }
445 if (pread(res->hr_localfd, buf, mapsize, METADATA_SIZE) !=
446 (ssize_t)mapsize) {
447 primary_exit(EX_NOINPUT, "Unable to read activemap");
448 }
449 activemap_copyin(res->hr_amp, buf, mapsize);
450 if (res->hr_resuid != 0)
451 return;
452 /*
453 * We're using provider for the first time, so we have to generate
454 * resource unique identifier and initialize local and remote counts.
455 */
456 arc4random_buf(&res->hr_resuid, sizeof(res->hr_resuid));
457 res->hr_primary_localcnt = 1;
458 res->hr_primary_remotecnt = 0;
459 if (metadata_write(res) < 0)
460 exit(EX_NOINPUT);
461}
462
463static bool
464init_remote(struct hast_resource *res, struct proto_conn **inp,
465 struct proto_conn **outp)
466{
467 struct proto_conn *in, *out;
468 struct nv *nvout, *nvin;
469 const unsigned char *token;
470 unsigned char *map;
471 const char *errmsg;
472 int32_t extentsize;
473 int64_t datasize;
474 uint32_t mapsize;
475 size_t size;
476
477 assert((inp == NULL && outp == NULL) || (inp != NULL && outp != NULL));
478
479 in = out = NULL;
480
481 /* Prepare outgoing connection with remote node. */
482 if (proto_client(res->hr_remoteaddr, &out) < 0) {
483 primary_exit(EX_TEMPFAIL, "Unable to create connection to %s",
484 res->hr_remoteaddr);
485 }
486 /* Try to connect, but accept failure. */
487 if (proto_connect(out) < 0) {
488 pjdlog_errno(LOG_WARNING, "Unable to connect to %s",
489 res->hr_remoteaddr);
490 goto close;
491 }
|
555 nvout = nv_alloc();
556 nv_add_string(nvout, res->hr_name, "resource");
557 nv_add_uint8_array(nvout, res->hr_token, sizeof(res->hr_token),
558 "token");
559 nv_add_uint64(nvout, res->hr_resuid, "resuid");
560 nv_add_uint64(nvout, res->hr_primary_localcnt, "localcnt");
561 nv_add_uint64(nvout, res->hr_primary_remotecnt, "remotecnt");
562 if (nv_error(nvout) != 0) {
563 pjdlog_common(LOG_WARNING, 0, nv_error(nvout),
564 "Unable to allocate header for connection with %s",
565 res->hr_remoteaddr);
566 nv_free(nvout);
567 goto close;
568 }
569 if (hast_proto_send(res, in, nvout, NULL, 0) < 0) {
570 pjdlog_errno(LOG_WARNING,
571 "Unable to send handshake header to %s",
572 res->hr_remoteaddr);
573 nv_free(nvout);
574 goto close;
575 }
576 nv_free(nvout);
577 if (hast_proto_recv_hdr(out, &nvin) < 0) {
578 pjdlog_errno(LOG_WARNING,
579 "Unable to receive handshake header from %s",
580 res->hr_remoteaddr);
581 goto close;
582 }
583 errmsg = nv_get_string(nvin, "errmsg");
584 if (errmsg != NULL) {
585 pjdlog_warning("%s", errmsg);
586 nv_free(nvin);
587 goto close;
588 }
589 datasize = nv_get_int64(nvin, "datasize");
590 if (datasize != res->hr_datasize) {
591 pjdlog_warning("Data size differs between nodes (local=%jd, remote=%jd).",
592 (intmax_t)res->hr_datasize, (intmax_t)datasize);
593 nv_free(nvin);
594 goto close;
595 }
596 extentsize = nv_get_int32(nvin, "extentsize");
597 if (extentsize != res->hr_extentsize) {
598 pjdlog_warning("Extent size differs between nodes (local=%zd, remote=%zd).",
599 (ssize_t)res->hr_extentsize, (ssize_t)extentsize);
600 nv_free(nvin);
601 goto close;
602 }
603 res->hr_secondary_localcnt = nv_get_uint64(nvin, "localcnt");
604 res->hr_secondary_remotecnt = nv_get_uint64(nvin, "remotecnt");
605 res->hr_syncsrc = nv_get_uint8(nvin, "syncsrc");
606 map = NULL;
607 mapsize = nv_get_uint32(nvin, "mapsize");
608 if (mapsize > 0) {
609 map = malloc(mapsize);
610 if (map == NULL) {
611 pjdlog_error("Unable to allocate memory for remote activemap (mapsize=%ju).",
612 (uintmax_t)mapsize);
613 nv_free(nvin);
614 goto close;
615 }
616 /*
617 * Remote node have some dirty extents on its own, lets
618 * download its activemap.
619 */
620 if (hast_proto_recv_data(res, out, nvin, map,
621 mapsize) < 0) {
622 pjdlog_errno(LOG_ERR,
623 "Unable to receive remote activemap");
624 nv_free(nvin);
625 free(map);
626 goto close;
627 }
628 /*
629 * Merge local and remote bitmaps.
630 */
631 activemap_merge(res->hr_amp, map, mapsize);
632 free(map);
633 /*
634 * Now that we merged bitmaps from both nodes, flush it to the
635 * disk before we start to synchronize.
636 */
637 (void)hast_activemap_flush(res);
638 }
639 pjdlog_info("Connected to %s.", res->hr_remoteaddr);
640 if (inp != NULL && outp != NULL) {
641 *inp = in;
642 *outp = out;
643 } else {
644 res->hr_remotein = in;
645 res->hr_remoteout = out;
646 }
647 return (true);
648close:
649 proto_close(out);
650 if (in != NULL)
651 proto_close(in);
652 return (false);
653}
654
655static void
656sync_start(void)
657{
658
659 mtx_lock(&sync_lock);
660 sync_inprogress = true;
661 mtx_unlock(&sync_lock);
662 cv_signal(&sync_cond);
663}
664
665static void
666init_ggate(struct hast_resource *res)
667{
668 struct g_gate_ctl_create ggiocreate;
669 struct g_gate_ctl_cancel ggiocancel;
670
671 /*
672 * We communicate with ggate via /dev/ggctl. Open it.
673 */
674 res->hr_ggatefd = open("/dev/" G_GATE_CTL_NAME, O_RDWR);
675 if (res->hr_ggatefd < 0)
676 primary_exit(EX_OSFILE, "Unable to open /dev/" G_GATE_CTL_NAME);
677 /*
678 * Create provider before trying to connect, as connection failure
679 * is not critical, but may take some time.
680 */
681 ggiocreate.gctl_version = G_GATE_VERSION;
682 ggiocreate.gctl_mediasize = res->hr_datasize;
683 ggiocreate.gctl_sectorsize = res->hr_local_sectorsize;
684 ggiocreate.gctl_flags = 0;
685 ggiocreate.gctl_maxcount = G_GATE_MAX_QUEUE_SIZE;
686 ggiocreate.gctl_timeout = 0;
687 ggiocreate.gctl_unit = G_GATE_NAME_GIVEN;
688 snprintf(ggiocreate.gctl_name, sizeof(ggiocreate.gctl_name), "hast/%s",
689 res->hr_provname);
690 bzero(ggiocreate.gctl_info, sizeof(ggiocreate.gctl_info));
691 if (ioctl(res->hr_ggatefd, G_GATE_CMD_CREATE, &ggiocreate) == 0) {
692 pjdlog_info("Device hast/%s created.", res->hr_provname);
693 res->hr_ggateunit = ggiocreate.gctl_unit;
694 return;
695 }
696 if (errno != EEXIST) {
697 primary_exit(EX_OSERR, "Unable to create hast/%s device",
698 res->hr_provname);
699 }
700 pjdlog_debug(1,
701 "Device hast/%s already exists, we will try to take it over.",
702 res->hr_provname);
703 /*
704 * If we received EEXIST, we assume that the process who created the
705 * provider died and didn't clean up. In that case we will start from
706 * where he left of.
707 */
708 ggiocancel.gctl_version = G_GATE_VERSION;
709 ggiocancel.gctl_unit = G_GATE_NAME_GIVEN;
710 snprintf(ggiocancel.gctl_name, sizeof(ggiocancel.gctl_name), "hast/%s",
711 res->hr_provname);
712 if (ioctl(res->hr_ggatefd, G_GATE_CMD_CANCEL, &ggiocancel) == 0) {
713 pjdlog_info("Device hast/%s recovered.", res->hr_provname);
714 res->hr_ggateunit = ggiocancel.gctl_unit;
715 return;
716 }
717 primary_exit(EX_OSERR, "Unable to take over hast/%s device",
718 res->hr_provname);
719}
720
721void
722hastd_primary(struct hast_resource *res)
723{
724 pthread_t td;
725 pid_t pid;
726 int error;
727
728 gres = res;
729
730 /*
731 * Create communication channel between parent and child.
732 */
733 if (proto_client("socketpair://", &res->hr_ctrl) < 0) {
734 KEEP_ERRNO((void)pidfile_remove(pfh));
735 primary_exit(EX_OSERR,
736 "Unable to create control sockets between parent and child");
737 }
738
739 pid = fork();
740 if (pid < 0) {
741 KEEP_ERRNO((void)pidfile_remove(pfh));
742 primary_exit(EX_TEMPFAIL, "Unable to fork");
743 }
744
745 if (pid > 0) {
746 /* This is parent. */
747 res->hr_workerpid = pid;
748 return;
749 }
750 (void)pidfile_close(pfh);
751
752 setproctitle("%s (primary)", res->hr_name);
753
754 init_local(res);
755 if (init_remote(res, NULL, NULL))
756 sync_start();
757 init_ggate(res);
758 init_environment(res);
759 error = pthread_create(&td, NULL, ggate_recv_thread, res);
760 assert(error == 0);
761 error = pthread_create(&td, NULL, local_send_thread, res);
762 assert(error == 0);
763 error = pthread_create(&td, NULL, remote_send_thread, res);
764 assert(error == 0);
765 error = pthread_create(&td, NULL, remote_recv_thread, res);
766 assert(error == 0);
767 error = pthread_create(&td, NULL, ggate_send_thread, res);
768 assert(error == 0);
769 error = pthread_create(&td, NULL, sync_thread, res);
770 assert(error == 0);
771 error = pthread_create(&td, NULL, ctrl_thread, res);
772 assert(error == 0);
773 (void)guard_thread(res);
774}
775
776static void
777reqlog(int loglevel, int debuglevel, struct g_gate_ctl_io *ggio, const char *fmt, ...)
778{
779 char msg[1024];
780 va_list ap;
781 int len;
782
783 va_start(ap, fmt);
784 len = vsnprintf(msg, sizeof(msg), fmt, ap);
785 va_end(ap);
786 if ((size_t)len < sizeof(msg)) {
787 switch (ggio->gctl_cmd) {
788 case BIO_READ:
789 (void)snprintf(msg + len, sizeof(msg) - len,
790 "READ(%ju, %ju).", (uintmax_t)ggio->gctl_offset,
791 (uintmax_t)ggio->gctl_length);
792 break;
793 case BIO_DELETE:
794 (void)snprintf(msg + len, sizeof(msg) - len,
795 "DELETE(%ju, %ju).", (uintmax_t)ggio->gctl_offset,
796 (uintmax_t)ggio->gctl_length);
797 break;
798 case BIO_FLUSH:
799 (void)snprintf(msg + len, sizeof(msg) - len, "FLUSH.");
800 break;
801 case BIO_WRITE:
802 (void)snprintf(msg + len, sizeof(msg) - len,
803 "WRITE(%ju, %ju).", (uintmax_t)ggio->gctl_offset,
804 (uintmax_t)ggio->gctl_length);
805 break;
806 default:
807 (void)snprintf(msg + len, sizeof(msg) - len,
808 "UNKNOWN(%u).", (unsigned int)ggio->gctl_cmd);
809 break;
810 }
811 }
812 pjdlog_common(loglevel, debuglevel, -1, "%s", msg);
813}
814
815static void
816remote_close(struct hast_resource *res, int ncomp)
817{
818
819 rw_wlock(&hio_remote_lock[ncomp]);
820 /*
821 * A race is possible between dropping rlock and acquiring wlock -
822 * another thread can close connection in-between.
823 */
824 if (!ISCONNECTED(res, ncomp)) {
825 assert(res->hr_remotein == NULL);
826 assert(res->hr_remoteout == NULL);
827 rw_unlock(&hio_remote_lock[ncomp]);
828 return;
829 }
830
831 assert(res->hr_remotein != NULL);
832 assert(res->hr_remoteout != NULL);
833
834 pjdlog_debug(2, "Closing old incoming connection to %s.",
835 res->hr_remoteaddr);
836 proto_close(res->hr_remotein);
837 res->hr_remotein = NULL;
838 pjdlog_debug(2, "Closing old outgoing connection to %s.",
839 res->hr_remoteaddr);
840 proto_close(res->hr_remoteout);
841 res->hr_remoteout = NULL;
842
843 rw_unlock(&hio_remote_lock[ncomp]);
844
845 /*
846 * Stop synchronization if in-progress.
847 */
848 mtx_lock(&sync_lock);
849 if (sync_inprogress)
850 sync_inprogress = false;
851 mtx_unlock(&sync_lock);
852
853 /*
854 * Wake up guard thread, so it can immediately start reconnect.
855 */
856 mtx_lock(&hio_guard_lock);
857 cv_signal(&hio_guard_cond);
858 mtx_unlock(&hio_guard_lock);
859}
860
861/*
862 * Thread receives ggate I/O requests from the kernel and passes them to
863 * appropriate threads:
864 * WRITE - always goes to both local_send and remote_send threads
865 * READ (when the block is up-to-date on local component) -
866 * only local_send thread
867 * READ (when the block isn't up-to-date on local component) -
868 * only remote_send thread
869 * DELETE - always goes to both local_send and remote_send threads
870 * FLUSH - always goes to both local_send and remote_send threads
871 */
872static void *
873ggate_recv_thread(void *arg)
874{
875 struct hast_resource *res = arg;
876 struct g_gate_ctl_io *ggio;
877 struct hio *hio;
878 unsigned int ii, ncomp, ncomps;
879 int error;
880
881 ncomps = HAST_NCOMPONENTS;
882
883 for (;;) {
884 pjdlog_debug(2, "ggate_recv: Taking free request.");
885 QUEUE_TAKE2(hio, free);
886 pjdlog_debug(2, "ggate_recv: (%p) Got free request.", hio);
887 ggio = &hio->hio_ggio;
888 ggio->gctl_unit = res->hr_ggateunit;
889 ggio->gctl_length = MAXPHYS;
890 ggio->gctl_error = 0;
891 pjdlog_debug(2,
892 "ggate_recv: (%p) Waiting for request from the kernel.",
893 hio);
894 if (ioctl(res->hr_ggatefd, G_GATE_CMD_START, ggio) < 0) {
895 if (sigexit_received)
896 pthread_exit(NULL);
897 primary_exit(EX_OSERR, "G_GATE_CMD_START failed");
898 }
899 error = ggio->gctl_error;
900 switch (error) {
901 case 0:
902 break;
903 case ECANCELED:
904 /* Exit gracefully. */
905 if (!sigexit_received) {
906 pjdlog_debug(2,
907 "ggate_recv: (%p) Received cancel from the kernel.",
908 hio);
909 pjdlog_info("Received cancel from the kernel, exiting.");
910 }
911 pthread_exit(NULL);
912 case ENOMEM:
913 /*
914 * Buffer too small? Impossible, we allocate MAXPHYS
915 * bytes - request can't be bigger than that.
916 */
917 /* FALLTHROUGH */
918 case ENXIO:
919 default:
920 primary_exitx(EX_OSERR, "G_GATE_CMD_START failed: %s.",
921 strerror(error));
922 }
923 for (ii = 0; ii < ncomps; ii++)
924 hio->hio_errors[ii] = EINVAL;
925 reqlog(LOG_DEBUG, 2, ggio,
926 "ggate_recv: (%p) Request received from the kernel: ",
927 hio);
928 /*
929 * Inform all components about new write request.
930 * For read request prefer local component unless the given
931 * range is out-of-date, then use remote component.
932 */
933 switch (ggio->gctl_cmd) {
934 case BIO_READ:
935 pjdlog_debug(2,
936 "ggate_recv: (%p) Moving request to the send queue.",
937 hio);
938 refcount_init(&hio->hio_countdown, 1);
939 mtx_lock(&metadata_lock);
940 if (res->hr_syncsrc == HAST_SYNCSRC_UNDEF ||
941 res->hr_syncsrc == HAST_SYNCSRC_PRIMARY) {
942 /*
943 * This range is up-to-date on local component,
944 * so handle request locally.
945 */
946 /* Local component is 0 for now. */
947 ncomp = 0;
948 } else /* if (res->hr_syncsrc ==
949 HAST_SYNCSRC_SECONDARY) */ {
950 assert(res->hr_syncsrc ==
951 HAST_SYNCSRC_SECONDARY);
952 /*
953 * This range is out-of-date on local component,
954 * so send request to the remote node.
955 */
956 /* Remote component is 1 for now. */
957 ncomp = 1;
958 }
959 mtx_unlock(&metadata_lock);
960 QUEUE_INSERT1(hio, send, ncomp);
961 break;
962 case BIO_WRITE:
963 for (;;) {
964 mtx_lock(&range_lock);
965 if (rangelock_islocked(range_sync,
966 ggio->gctl_offset, ggio->gctl_length)) {
967 pjdlog_debug(2,
968 "regular: Range offset=%jd length=%zu locked.",
969 (intmax_t)ggio->gctl_offset,
970 (size_t)ggio->gctl_length);
971 range_regular_wait = true;
972 cv_wait(&range_regular_cond, &range_lock);
973 range_regular_wait = false;
974 mtx_unlock(&range_lock);
975 continue;
976 }
977 if (rangelock_add(range_regular,
978 ggio->gctl_offset, ggio->gctl_length) < 0) {
979 mtx_unlock(&range_lock);
980 pjdlog_debug(2,
981 "regular: Range offset=%jd length=%zu is already locked, waiting.",
982 (intmax_t)ggio->gctl_offset,
983 (size_t)ggio->gctl_length);
984 sleep(1);
985 continue;
986 }
987 mtx_unlock(&range_lock);
988 break;
989 }
990 mtx_lock(&res->hr_amp_lock);
991 if (activemap_write_start(res->hr_amp,
992 ggio->gctl_offset, ggio->gctl_length)) {
993 (void)hast_activemap_flush(res);
994 }
995 mtx_unlock(&res->hr_amp_lock);
996 /* FALLTHROUGH */
997 case BIO_DELETE:
998 case BIO_FLUSH:
999 pjdlog_debug(2,
1000 "ggate_recv: (%p) Moving request to the send queues.",
1001 hio);
1002 refcount_init(&hio->hio_countdown, ncomps);
1003 for (ii = 0; ii < ncomps; ii++)
1004 QUEUE_INSERT1(hio, send, ii);
1005 break;
1006 }
1007 }
1008 /* NOTREACHED */
1009 return (NULL);
1010}
1011
1012/*
1013 * Thread reads from or writes to local component.
1014 * If local read fails, it redirects it to remote_send thread.
1015 */
1016static void *
1017local_send_thread(void *arg)
1018{
1019 struct hast_resource *res = arg;
1020 struct g_gate_ctl_io *ggio;
1021 struct hio *hio;
1022 unsigned int ncomp, rncomp;
1023 ssize_t ret;
1024
1025 /* Local component is 0 for now. */
1026 ncomp = 0;
1027 /* Remote component is 1 for now. */
1028 rncomp = 1;
1029
1030 for (;;) {
1031 pjdlog_debug(2, "local_send: Taking request.");
1032 QUEUE_TAKE1(hio, send, ncomp);
1033 pjdlog_debug(2, "local_send: (%p) Got request.", hio);
1034 ggio = &hio->hio_ggio;
1035 switch (ggio->gctl_cmd) {
1036 case BIO_READ:
1037 ret = pread(res->hr_localfd, ggio->gctl_data,
1038 ggio->gctl_length,
1039 ggio->gctl_offset + res->hr_localoff);
1040 if (ret == ggio->gctl_length)
1041 hio->hio_errors[ncomp] = 0;
1042 else {
1043 /*
1044 * If READ failed, try to read from remote node.
1045 */
1046 QUEUE_INSERT1(hio, send, rncomp);
1047 continue;
1048 }
1049 break;
1050 case BIO_WRITE:
1051 ret = pwrite(res->hr_localfd, ggio->gctl_data,
1052 ggio->gctl_length,
1053 ggio->gctl_offset + res->hr_localoff);
1054 if (ret < 0)
1055 hio->hio_errors[ncomp] = errno;
1056 else if (ret != ggio->gctl_length)
1057 hio->hio_errors[ncomp] = EIO;
1058 else
1059 hio->hio_errors[ncomp] = 0;
1060 break;
1061 case BIO_DELETE:
1062 ret = g_delete(res->hr_localfd,
1063 ggio->gctl_offset + res->hr_localoff,
1064 ggio->gctl_length);
1065 if (ret < 0)
1066 hio->hio_errors[ncomp] = errno;
1067 else
1068 hio->hio_errors[ncomp] = 0;
1069 break;
1070 case BIO_FLUSH:
1071 ret = g_flush(res->hr_localfd);
1072 if (ret < 0)
1073 hio->hio_errors[ncomp] = errno;
1074 else
1075 hio->hio_errors[ncomp] = 0;
1076 break;
1077 }
1078 if (refcount_release(&hio->hio_countdown)) {
1079 if (ISSYNCREQ(hio)) {
1080 mtx_lock(&sync_lock);
1081 SYNCREQDONE(hio);
1082 mtx_unlock(&sync_lock);
1083 cv_signal(&sync_cond);
1084 } else {
1085 pjdlog_debug(2,
1086 "local_send: (%p) Moving request to the done queue.",
1087 hio);
1088 QUEUE_INSERT2(hio, done);
1089 }
1090 }
1091 }
1092 /* NOTREACHED */
1093 return (NULL);
1094}
1095
1096/*
1097 * Thread sends request to secondary node.
1098 */
1099static void *
1100remote_send_thread(void *arg)
1101{
1102 struct hast_resource *res = arg;
1103 struct g_gate_ctl_io *ggio;
1104 struct hio *hio;
1105 struct nv *nv;
1106 unsigned int ncomp;
1107 bool wakeup;
1108 uint64_t offset, length;
1109 uint8_t cmd;
1110 void *data;
1111
1112 /* Remote component is 1 for now. */
1113 ncomp = 1;
1114
1115 for (;;) {
1116 pjdlog_debug(2, "remote_send: Taking request.");
1117 QUEUE_TAKE1(hio, send, ncomp);
1118 pjdlog_debug(2, "remote_send: (%p) Got request.", hio);
1119 ggio = &hio->hio_ggio;
1120 switch (ggio->gctl_cmd) {
1121 case BIO_READ:
1122 cmd = HIO_READ;
1123 data = NULL;
1124 offset = ggio->gctl_offset;
1125 length = ggio->gctl_length;
1126 break;
1127 case BIO_WRITE:
1128 cmd = HIO_WRITE;
1129 data = ggio->gctl_data;
1130 offset = ggio->gctl_offset;
1131 length = ggio->gctl_length;
1132 break;
1133 case BIO_DELETE:
1134 cmd = HIO_DELETE;
1135 data = NULL;
1136 offset = ggio->gctl_offset;
1137 length = ggio->gctl_length;
1138 break;
1139 case BIO_FLUSH:
1140 cmd = HIO_FLUSH;
1141 data = NULL;
1142 offset = 0;
1143 length = 0;
1144 break;
1145 default:
1146 assert(!"invalid condition");
1147 abort();
1148 }
1149 nv = nv_alloc();
1150 nv_add_uint8(nv, cmd, "cmd");
1151 nv_add_uint64(nv, (uint64_t)ggio->gctl_seq, "seq");
1152 nv_add_uint64(nv, offset, "offset");
1153 nv_add_uint64(nv, length, "length");
1154 if (nv_error(nv) != 0) {
1155 hio->hio_errors[ncomp] = nv_error(nv);
1156 pjdlog_debug(2,
1157 "remote_send: (%p) Unable to prepare header to send.",
1158 hio);
1159 reqlog(LOG_ERR, 0, ggio,
1160 "Unable to prepare header to send (%s): ",
1161 strerror(nv_error(nv)));
1162 /* Move failed request immediately to the done queue. */
1163 goto done_queue;
1164 }
1165 pjdlog_debug(2,
1166 "remote_send: (%p) Moving request to the recv queue.",
1167 hio);
1168 /*
1169 * Protect connection from disappearing.
1170 */
1171 rw_rlock(&hio_remote_lock[ncomp]);
1172 if (!ISCONNECTED(res, ncomp)) {
1173 rw_unlock(&hio_remote_lock[ncomp]);
1174 hio->hio_errors[ncomp] = ENOTCONN;
1175 goto done_queue;
1176 }
1177 /*
1178 * Move the request to recv queue before sending it, because
1179 * in different order we can get reply before we move request
1180 * to recv queue.
1181 */
1182 mtx_lock(&hio_recv_list_lock[ncomp]);
1183 wakeup = TAILQ_EMPTY(&hio_recv_list[ncomp]);
1184 TAILQ_INSERT_TAIL(&hio_recv_list[ncomp], hio, hio_next[ncomp]);
1185 mtx_unlock(&hio_recv_list_lock[ncomp]);
1186 if (hast_proto_send(res, res->hr_remoteout, nv, data,
1187 data != NULL ? length : 0) < 0) {
1188 hio->hio_errors[ncomp] = errno;
1189 rw_unlock(&hio_remote_lock[ncomp]);
1190 remote_close(res, ncomp);
1191 pjdlog_debug(2,
1192 "remote_send: (%p) Unable to send request.", hio);
1193 reqlog(LOG_ERR, 0, ggio,
1194 "Unable to send request (%s): ",
1195 strerror(hio->hio_errors[ncomp]));
1196 /*
1197 * Take request back from the receive queue and move
1198 * it immediately to the done queue.
1199 */
1200 mtx_lock(&hio_recv_list_lock[ncomp]);
1201 TAILQ_REMOVE(&hio_recv_list[ncomp], hio, hio_next[ncomp]);
1202 mtx_unlock(&hio_recv_list_lock[ncomp]);
1203 goto done_queue;
1204 }
1205 rw_unlock(&hio_remote_lock[ncomp]);
1206 nv_free(nv);
1207 if (wakeup)
1208 cv_signal(&hio_recv_list_cond[ncomp]);
1209 continue;
1210done_queue:
1211 nv_free(nv);
1212 if (ISSYNCREQ(hio)) {
1213 if (!refcount_release(&hio->hio_countdown))
1214 continue;
1215 mtx_lock(&sync_lock);
1216 SYNCREQDONE(hio);
1217 mtx_unlock(&sync_lock);
1218 cv_signal(&sync_cond);
1219 continue;
1220 }
1221 if (ggio->gctl_cmd == BIO_WRITE) {
1222 mtx_lock(&res->hr_amp_lock);
1223 if (activemap_need_sync(res->hr_amp, ggio->gctl_offset,
1224 ggio->gctl_length)) {
1225 (void)hast_activemap_flush(res);
1226 }
1227 mtx_unlock(&res->hr_amp_lock);
1228 }
1229 if (!refcount_release(&hio->hio_countdown))
1230 continue;
1231 pjdlog_debug(2,
1232 "remote_send: (%p) Moving request to the done queue.",
1233 hio);
1234 QUEUE_INSERT2(hio, done);
1235 }
1236 /* NOTREACHED */
1237 return (NULL);
1238}
1239
1240/*
1241 * Thread receives answer from secondary node and passes it to ggate_send
1242 * thread.
1243 */
1244static void *
1245remote_recv_thread(void *arg)
1246{
1247 struct hast_resource *res = arg;
1248 struct g_gate_ctl_io *ggio;
1249 struct hio *hio;
1250 struct nv *nv;
1251 unsigned int ncomp;
1252 uint64_t seq;
1253 int error;
1254
1255 /* Remote component is 1 for now. */
1256 ncomp = 1;
1257
1258 for (;;) {
1259 /* Wait until there is anything to receive. */
1260 mtx_lock(&hio_recv_list_lock[ncomp]);
1261 while (TAILQ_EMPTY(&hio_recv_list[ncomp])) {
1262 pjdlog_debug(2, "remote_recv: No requests, waiting.");
1263 cv_wait(&hio_recv_list_cond[ncomp],
1264 &hio_recv_list_lock[ncomp]);
1265 }
1266 mtx_unlock(&hio_recv_list_lock[ncomp]);
1267 rw_rlock(&hio_remote_lock[ncomp]);
1268 if (!ISCONNECTED(res, ncomp)) {
1269 rw_unlock(&hio_remote_lock[ncomp]);
1270 /*
1271 * Connection is dead, so move all pending requests to
1272 * the done queue (one-by-one).
1273 */
1274 mtx_lock(&hio_recv_list_lock[ncomp]);
1275 hio = TAILQ_FIRST(&hio_recv_list[ncomp]);
1276 assert(hio != NULL);
1277 TAILQ_REMOVE(&hio_recv_list[ncomp], hio,
1278 hio_next[ncomp]);
1279 mtx_unlock(&hio_recv_list_lock[ncomp]);
1280 goto done_queue;
1281 }
1282 if (hast_proto_recv_hdr(res->hr_remotein, &nv) < 0) {
1283 pjdlog_errno(LOG_ERR,
1284 "Unable to receive reply header");
1285 rw_unlock(&hio_remote_lock[ncomp]);
1286 remote_close(res, ncomp);
1287 continue;
1288 }
1289 rw_unlock(&hio_remote_lock[ncomp]);
1290 seq = nv_get_uint64(nv, "seq");
1291 if (seq == 0) {
1292 pjdlog_error("Header contains no 'seq' field.");
1293 nv_free(nv);
1294 continue;
1295 }
1296 mtx_lock(&hio_recv_list_lock[ncomp]);
1297 TAILQ_FOREACH(hio, &hio_recv_list[ncomp], hio_next[ncomp]) {
1298 if (hio->hio_ggio.gctl_seq == seq) {
1299 TAILQ_REMOVE(&hio_recv_list[ncomp], hio,
1300 hio_next[ncomp]);
1301 break;
1302 }
1303 }
1304 mtx_unlock(&hio_recv_list_lock[ncomp]);
1305 if (hio == NULL) {
1306 pjdlog_error("Found no request matching received 'seq' field (%ju).",
1307 (uintmax_t)seq);
1308 nv_free(nv);
1309 continue;
1310 }
1311 error = nv_get_int16(nv, "error");
1312 if (error != 0) {
1313 /* Request failed on remote side. */
1314 hio->hio_errors[ncomp] = 0;
1315 nv_free(nv);
1316 goto done_queue;
1317 }
1318 ggio = &hio->hio_ggio;
1319 switch (ggio->gctl_cmd) {
1320 case BIO_READ:
1321 rw_rlock(&hio_remote_lock[ncomp]);
1322 if (!ISCONNECTED(res, ncomp)) {
1323 rw_unlock(&hio_remote_lock[ncomp]);
1324 nv_free(nv);
1325 goto done_queue;
1326 }
1327 if (hast_proto_recv_data(res, res->hr_remotein, nv,
1328 ggio->gctl_data, ggio->gctl_length) < 0) {
1329 hio->hio_errors[ncomp] = errno;
1330 pjdlog_errno(LOG_ERR,
1331 "Unable to receive reply data");
1332 rw_unlock(&hio_remote_lock[ncomp]);
1333 nv_free(nv);
1334 remote_close(res, ncomp);
1335 goto done_queue;
1336 }
1337 rw_unlock(&hio_remote_lock[ncomp]);
1338 break;
1339 case BIO_WRITE:
1340 case BIO_DELETE:
1341 case BIO_FLUSH:
1342 break;
1343 default:
1344 assert(!"invalid condition");
1345 abort();
1346 }
1347 hio->hio_errors[ncomp] = 0;
1348 nv_free(nv);
1349done_queue:
1350 if (refcount_release(&hio->hio_countdown)) {
1351 if (ISSYNCREQ(hio)) {
1352 mtx_lock(&sync_lock);
1353 SYNCREQDONE(hio);
1354 mtx_unlock(&sync_lock);
1355 cv_signal(&sync_cond);
1356 } else {
1357 pjdlog_debug(2,
1358 "remote_recv: (%p) Moving request to the done queue.",
1359 hio);
1360 QUEUE_INSERT2(hio, done);
1361 }
1362 }
1363 }
1364 /* NOTREACHED */
1365 return (NULL);
1366}
1367
1368/*
1369 * Thread sends answer to the kernel.
1370 */
1371static void *
1372ggate_send_thread(void *arg)
1373{
1374 struct hast_resource *res = arg;
1375 struct g_gate_ctl_io *ggio;
1376 struct hio *hio;
1377 unsigned int ii, ncomp, ncomps;
1378
1379 ncomps = HAST_NCOMPONENTS;
1380
1381 for (;;) {
1382 pjdlog_debug(2, "ggate_send: Taking request.");
1383 QUEUE_TAKE2(hio, done);
1384 pjdlog_debug(2, "ggate_send: (%p) Got request.", hio);
1385 ggio = &hio->hio_ggio;
1386 for (ii = 0; ii < ncomps; ii++) {
1387 if (hio->hio_errors[ii] == 0) {
1388 /*
1389 * One successful request is enough to declare
1390 * success.
1391 */
1392 ggio->gctl_error = 0;
1393 break;
1394 }
1395 }
1396 if (ii == ncomps) {
1397 /*
1398 * None of the requests were successful.
1399 * Use first error.
1400 */
1401 ggio->gctl_error = hio->hio_errors[0];
1402 }
1403 if (ggio->gctl_error == 0 && ggio->gctl_cmd == BIO_WRITE) {
1404 mtx_lock(&res->hr_amp_lock);
1405 activemap_write_complete(res->hr_amp,
1406 ggio->gctl_offset, ggio->gctl_length);
1407 mtx_unlock(&res->hr_amp_lock);
1408 }
1409 if (ggio->gctl_cmd == BIO_WRITE) {
1410 /*
1411 * Unlock range we locked.
1412 */
1413 mtx_lock(&range_lock);
1414 rangelock_del(range_regular, ggio->gctl_offset,
1415 ggio->gctl_length);
1416 if (range_sync_wait)
1417 cv_signal(&range_sync_cond);
1418 mtx_unlock(&range_lock);
1419 /*
1420 * Bump local count if this is first write after
1421 * connection failure with remote node.
1422 */
1423 ncomp = 1;
1424 rw_rlock(&hio_remote_lock[ncomp]);
1425 if (!ISCONNECTED(res, ncomp)) {
1426 mtx_lock(&metadata_lock);
1427 if (res->hr_primary_localcnt ==
1428 res->hr_secondary_remotecnt) {
1429 res->hr_primary_localcnt++;
1430 pjdlog_debug(1,
1431 "Increasing localcnt to %ju.",
1432 (uintmax_t)res->hr_primary_localcnt);
1433 (void)metadata_write(res);
1434 }
1435 mtx_unlock(&metadata_lock);
1436 }
1437 rw_unlock(&hio_remote_lock[ncomp]);
1438 }
1439 if (ioctl(res->hr_ggatefd, G_GATE_CMD_DONE, ggio) < 0)
1440 primary_exit(EX_OSERR, "G_GATE_CMD_DONE failed");
1441 pjdlog_debug(2,
1442 "ggate_send: (%p) Moving request to the free queue.", hio);
1443 QUEUE_INSERT2(hio, free);
1444 }
1445 /* NOTREACHED */
1446 return (NULL);
1447}
1448
1449/*
1450 * Thread synchronize local and remote components.
1451 */
1452static void *
1453sync_thread(void *arg __unused)
1454{
1455 struct hast_resource *res = arg;
1456 struct hio *hio;
1457 struct g_gate_ctl_io *ggio;
1458 unsigned int ii, ncomp, ncomps;
1459 off_t offset, length, synced;
1460 bool dorewind;
1461 int syncext;
1462
1463 ncomps = HAST_NCOMPONENTS;
1464 dorewind = true;
1465 synced = 0;
1466
1467 for (;;) {
1468 mtx_lock(&sync_lock);
1469 while (!sync_inprogress) {
1470 dorewind = true;
1471 synced = 0;
1472 cv_wait(&sync_cond, &sync_lock);
1473 }
1474 mtx_unlock(&sync_lock);
1475 /*
1476 * Obtain offset at which we should synchronize.
1477 * Rewind synchronization if needed.
1478 */
1479 mtx_lock(&res->hr_amp_lock);
1480 if (dorewind)
1481 activemap_sync_rewind(res->hr_amp);
1482 offset = activemap_sync_offset(res->hr_amp, &length, &syncext);
1483 if (syncext != -1) {
1484 /*
1485 * We synchronized entire syncext extent, we can mark
1486 * it as clean now.
1487 */
1488 if (activemap_extent_complete(res->hr_amp, syncext))
1489 (void)hast_activemap_flush(res);
1490 }
1491 mtx_unlock(&res->hr_amp_lock);
1492 if (dorewind) {
1493 dorewind = false;
1494 if (offset < 0)
1495 pjdlog_info("Nodes are in sync.");
1496 else {
1497 pjdlog_info("Synchronization started. %ju bytes to go.",
1498 (uintmax_t)(res->hr_extentsize *
1499 activemap_ndirty(res->hr_amp)));
1500 }
1501 }
1502 if (offset < 0) {
1503 mtx_lock(&sync_lock);
1504 sync_inprogress = false;
1505 mtx_unlock(&sync_lock);
1506 pjdlog_debug(1, "Nothing to synchronize.");
1507 /*
1508 * Synchronization complete, make both localcnt and
1509 * remotecnt equal.
1510 */
1511 ncomp = 1;
1512 rw_rlock(&hio_remote_lock[ncomp]);
1513 if (ISCONNECTED(res, ncomp)) {
1514 if (synced > 0) {
1515 pjdlog_info("Synchronization complete. "
1516 "%jd bytes synchronized.",
1517 (intmax_t)synced);
1518 }
1519 mtx_lock(&metadata_lock);
1520 res->hr_syncsrc = HAST_SYNCSRC_UNDEF;
1521 res->hr_primary_localcnt =
1522 res->hr_secondary_localcnt;
1523 res->hr_primary_remotecnt =
1524 res->hr_secondary_remotecnt;
1525 pjdlog_debug(1,
1526 "Setting localcnt to %ju and remotecnt to %ju.",
1527 (uintmax_t)res->hr_primary_localcnt,
1528 (uintmax_t)res->hr_secondary_localcnt);
1529 (void)metadata_write(res);
1530 mtx_unlock(&metadata_lock);
1531 } else if (synced > 0) {
1532 pjdlog_info("Synchronization interrupted. "
1533 "%jd bytes synchronized so far.",
1534 (intmax_t)synced);
1535 }
1536 rw_unlock(&hio_remote_lock[ncomp]);
1537 continue;
1538 }
1539 pjdlog_debug(2, "sync: Taking free request.");
1540 QUEUE_TAKE2(hio, free);
1541 pjdlog_debug(2, "sync: (%p) Got free request.", hio);
1542 /*
1543 * Lock the range we are going to synchronize. We don't want
1544 * race where someone writes between our read and write.
1545 */
1546 for (;;) {
1547 mtx_lock(&range_lock);
1548 if (rangelock_islocked(range_regular, offset, length)) {
1549 pjdlog_debug(2,
1550 "sync: Range offset=%jd length=%jd locked.",
1551 (intmax_t)offset, (intmax_t)length);
1552 range_sync_wait = true;
1553 cv_wait(&range_sync_cond, &range_lock);
1554 range_sync_wait = false;
1555 mtx_unlock(&range_lock);
1556 continue;
1557 }
1558 if (rangelock_add(range_sync, offset, length) < 0) {
1559 mtx_unlock(&range_lock);
1560 pjdlog_debug(2,
1561 "sync: Range offset=%jd length=%jd is already locked, waiting.",
1562 (intmax_t)offset, (intmax_t)length);
1563 sleep(1);
1564 continue;
1565 }
1566 mtx_unlock(&range_lock);
1567 break;
1568 }
1569 /*
1570 * First read the data from synchronization source.
1571 */
1572 SYNCREQ(hio);
1573 ggio = &hio->hio_ggio;
1574 ggio->gctl_cmd = BIO_READ;
1575 ggio->gctl_offset = offset;
1576 ggio->gctl_length = length;
1577 ggio->gctl_error = 0;
1578 for (ii = 0; ii < ncomps; ii++)
1579 hio->hio_errors[ii] = EINVAL;
1580 reqlog(LOG_DEBUG, 2, ggio, "sync: (%p) Sending sync request: ",
1581 hio);
1582 pjdlog_debug(2, "sync: (%p) Moving request to the send queue.",
1583 hio);
1584 mtx_lock(&metadata_lock);
1585 if (res->hr_syncsrc == HAST_SYNCSRC_PRIMARY) {
1586 /*
1587 * This range is up-to-date on local component,
1588 * so handle request locally.
1589 */
1590 /* Local component is 0 for now. */
1591 ncomp = 0;
1592 } else /* if (res->hr_syncsrc == HAST_SYNCSRC_SECONDARY) */ {
1593 assert(res->hr_syncsrc == HAST_SYNCSRC_SECONDARY);
1594 /*
1595 * This range is out-of-date on local component,
1596 * so send request to the remote node.
1597 */
1598 /* Remote component is 1 for now. */
1599 ncomp = 1;
1600 }
1601 mtx_unlock(&metadata_lock);
1602 refcount_init(&hio->hio_countdown, 1);
1603 QUEUE_INSERT1(hio, send, ncomp);
1604
1605 /*
1606 * Let's wait for READ to finish.
1607 */
1608 mtx_lock(&sync_lock);
1609 while (!ISSYNCREQDONE(hio))
1610 cv_wait(&sync_cond, &sync_lock);
1611 mtx_unlock(&sync_lock);
1612
1613 if (hio->hio_errors[ncomp] != 0) {
1614 pjdlog_error("Unable to read synchronization data: %s.",
1615 strerror(hio->hio_errors[ncomp]));
1616 goto free_queue;
1617 }
1618
1619 /*
1620 * We read the data from synchronization source, now write it
1621 * to synchronization target.
1622 */
1623 SYNCREQ(hio);
1624 ggio->gctl_cmd = BIO_WRITE;
1625 for (ii = 0; ii < ncomps; ii++)
1626 hio->hio_errors[ii] = EINVAL;
1627 reqlog(LOG_DEBUG, 2, ggio, "sync: (%p) Sending sync request: ",
1628 hio);
1629 pjdlog_debug(2, "sync: (%p) Moving request to the send queue.",
1630 hio);
1631 mtx_lock(&metadata_lock);
1632 if (res->hr_syncsrc == HAST_SYNCSRC_PRIMARY) {
1633 /*
1634 * This range is up-to-date on local component,
1635 * so we update remote component.
1636 */
1637 /* Remote component is 1 for now. */
1638 ncomp = 1;
1639 } else /* if (res->hr_syncsrc == HAST_SYNCSRC_SECONDARY) */ {
1640 assert(res->hr_syncsrc == HAST_SYNCSRC_SECONDARY);
1641 /*
1642 * This range is out-of-date on local component,
1643 * so we update it.
1644 */
1645 /* Local component is 0 for now. */
1646 ncomp = 0;
1647 }
1648 mtx_unlock(&metadata_lock);
1649
1650 pjdlog_debug(2, "sync: (%p) Moving request to the send queues.",
1651 hio);
1652 refcount_init(&hio->hio_countdown, 1);
1653 QUEUE_INSERT1(hio, send, ncomp);
1654
1655 /*
1656 * Let's wait for WRITE to finish.
1657 */
1658 mtx_lock(&sync_lock);
1659 while (!ISSYNCREQDONE(hio))
1660 cv_wait(&sync_cond, &sync_lock);
1661 mtx_unlock(&sync_lock);
1662
1663 if (hio->hio_errors[ncomp] != 0) {
1664 pjdlog_error("Unable to write synchronization data: %s.",
1665 strerror(hio->hio_errors[ncomp]));
1666 goto free_queue;
1667 }
1668free_queue:
1669 mtx_lock(&range_lock);
1670 rangelock_del(range_sync, offset, length);
1671 if (range_regular_wait)
1672 cv_signal(&range_regular_cond);
1673 mtx_unlock(&range_lock);
1674
1675 synced += length;
1676
1677 pjdlog_debug(2, "sync: (%p) Moving request to the free queue.",
1678 hio);
1679 QUEUE_INSERT2(hio, free);
1680 }
1681 /* NOTREACHED */
1682 return (NULL);
1683}
1684
1685static void
1686sighandler(int sig)
1687{
1688 bool unlock;
1689
1690 switch (sig) {
1691 case SIGINT:
1692 case SIGTERM:
1693 sigexit_received = true;
1694 break;
1695 default:
1696 assert(!"invalid condition");
1697 }
1698 /*
1699 * XXX: Racy, but if we cannot obtain hio_guard_lock here, we don't
1700 * want to risk deadlock.
1701 */
1702 unlock = mtx_trylock(&hio_guard_lock);
1703 cv_signal(&hio_guard_cond);
1704 if (unlock)
1705 mtx_unlock(&hio_guard_lock);
1706}
1707
1708/*
1709 * Thread guards remote connections and reconnects when needed, handles
1710 * signals, etc.
1711 */
1712static void *
1713guard_thread(void *arg)
1714{
1715 struct hast_resource *res = arg;
1716 struct proto_conn *in, *out;
1717 unsigned int ii, ncomps;
1718 int timeout;
1719
1720 ncomps = HAST_NCOMPONENTS;
1721 /* The is only one remote component for now. */
1722#define ISREMOTE(no) ((no) == 1)
1723
1724 for (;;) {
1725 if (sigexit_received) {
1726 primary_exitx(EX_OK,
1727 "Termination signal received, exiting.");
1728 }
1729 /*
1730 * If all the connection will be fine, we will sleep until
1731 * someone wakes us up.
1732 * If any of the connections will be broken and we won't be
1733 * able to connect, we will sleep only for RECONNECT_SLEEP
1734 * seconds so we can retry soon.
1735 */
1736 timeout = 0;
1737 pjdlog_debug(2, "remote_guard: Checking connections.");
1738 mtx_lock(&hio_guard_lock);
1739 for (ii = 0; ii < ncomps; ii++) {
1740 if (!ISREMOTE(ii))
1741 continue;
1742 rw_rlock(&hio_remote_lock[ii]);
1743 if (ISCONNECTED(res, ii)) {
1744 assert(res->hr_remotein != NULL);
1745 assert(res->hr_remoteout != NULL);
1746 rw_unlock(&hio_remote_lock[ii]);
1747 pjdlog_debug(2,
1748 "remote_guard: Connection to %s is ok.",
1749 res->hr_remoteaddr);
1750 } else {
1751 assert(res->hr_remotein == NULL);
1752 assert(res->hr_remoteout == NULL);
1753 /*
1754 * Upgrade the lock. It doesn't have to be
1755 * atomic as no other thread can change
1756 * connection status from disconnected to
1757 * connected.
1758 */
1759 rw_unlock(&hio_remote_lock[ii]);
1760 pjdlog_debug(2,
1761 "remote_guard: Reconnecting to %s.",
1762 res->hr_remoteaddr);
1763 in = out = NULL;
1764 if (init_remote(res, &in, &out)) {
1765 rw_wlock(&hio_remote_lock[ii]);
1766 assert(res->hr_remotein == NULL);
1767 assert(res->hr_remoteout == NULL);
1768 assert(in != NULL && out != NULL);
1769 res->hr_remotein = in;
1770 res->hr_remoteout = out;
1771 rw_unlock(&hio_remote_lock[ii]);
1772 pjdlog_info("Successfully reconnected to %s.",
1773 res->hr_remoteaddr);
1774 sync_start();
1775 } else {
1776 /* Both connections should be NULL. */
1777 assert(res->hr_remotein == NULL);
1778 assert(res->hr_remoteout == NULL);
1779 assert(in == NULL && out == NULL);
1780 pjdlog_debug(2,
1781 "remote_guard: Reconnect to %s failed.",
1782 res->hr_remoteaddr);
1783 timeout = RECONNECT_SLEEP;
1784 }
1785 }
1786 }
1787 (void)cv_timedwait(&hio_guard_cond, &hio_guard_lock, timeout);
1788 mtx_unlock(&hio_guard_lock);
1789 }
1790#undef ISREMOTE
1791 /* NOTREACHED */
1792 return (NULL);
1793}
| 561 nvout = nv_alloc();
562 nv_add_string(nvout, res->hr_name, "resource");
563 nv_add_uint8_array(nvout, res->hr_token, sizeof(res->hr_token),
564 "token");
565 nv_add_uint64(nvout, res->hr_resuid, "resuid");
566 nv_add_uint64(nvout, res->hr_primary_localcnt, "localcnt");
567 nv_add_uint64(nvout, res->hr_primary_remotecnt, "remotecnt");
568 if (nv_error(nvout) != 0) {
569 pjdlog_common(LOG_WARNING, 0, nv_error(nvout),
570 "Unable to allocate header for connection with %s",
571 res->hr_remoteaddr);
572 nv_free(nvout);
573 goto close;
574 }
575 if (hast_proto_send(res, in, nvout, NULL, 0) < 0) {
576 pjdlog_errno(LOG_WARNING,
577 "Unable to send handshake header to %s",
578 res->hr_remoteaddr);
579 nv_free(nvout);
580 goto close;
581 }
582 nv_free(nvout);
583 if (hast_proto_recv_hdr(out, &nvin) < 0) {
584 pjdlog_errno(LOG_WARNING,
585 "Unable to receive handshake header from %s",
586 res->hr_remoteaddr);
587 goto close;
588 }
589 errmsg = nv_get_string(nvin, "errmsg");
590 if (errmsg != NULL) {
591 pjdlog_warning("%s", errmsg);
592 nv_free(nvin);
593 goto close;
594 }
595 datasize = nv_get_int64(nvin, "datasize");
596 if (datasize != res->hr_datasize) {
597 pjdlog_warning("Data size differs between nodes (local=%jd, remote=%jd).",
598 (intmax_t)res->hr_datasize, (intmax_t)datasize);
599 nv_free(nvin);
600 goto close;
601 }
602 extentsize = nv_get_int32(nvin, "extentsize");
603 if (extentsize != res->hr_extentsize) {
604 pjdlog_warning("Extent size differs between nodes (local=%zd, remote=%zd).",
605 (ssize_t)res->hr_extentsize, (ssize_t)extentsize);
606 nv_free(nvin);
607 goto close;
608 }
609 res->hr_secondary_localcnt = nv_get_uint64(nvin, "localcnt");
610 res->hr_secondary_remotecnt = nv_get_uint64(nvin, "remotecnt");
611 res->hr_syncsrc = nv_get_uint8(nvin, "syncsrc");
612 map = NULL;
613 mapsize = nv_get_uint32(nvin, "mapsize");
614 if (mapsize > 0) {
615 map = malloc(mapsize);
616 if (map == NULL) {
617 pjdlog_error("Unable to allocate memory for remote activemap (mapsize=%ju).",
618 (uintmax_t)mapsize);
619 nv_free(nvin);
620 goto close;
621 }
622 /*
623 * Remote node have some dirty extents on its own, lets
624 * download its activemap.
625 */
626 if (hast_proto_recv_data(res, out, nvin, map,
627 mapsize) < 0) {
628 pjdlog_errno(LOG_ERR,
629 "Unable to receive remote activemap");
630 nv_free(nvin);
631 free(map);
632 goto close;
633 }
634 /*
635 * Merge local and remote bitmaps.
636 */
637 activemap_merge(res->hr_amp, map, mapsize);
638 free(map);
639 /*
640 * Now that we merged bitmaps from both nodes, flush it to the
641 * disk before we start to synchronize.
642 */
643 (void)hast_activemap_flush(res);
644 }
645 pjdlog_info("Connected to %s.", res->hr_remoteaddr);
646 if (inp != NULL && outp != NULL) {
647 *inp = in;
648 *outp = out;
649 } else {
650 res->hr_remotein = in;
651 res->hr_remoteout = out;
652 }
653 return (true);
654close:
655 proto_close(out);
656 if (in != NULL)
657 proto_close(in);
658 return (false);
659}
660
661static void
662sync_start(void)
663{
664
665 mtx_lock(&sync_lock);
666 sync_inprogress = true;
667 mtx_unlock(&sync_lock);
668 cv_signal(&sync_cond);
669}
670
671static void
672init_ggate(struct hast_resource *res)
673{
674 struct g_gate_ctl_create ggiocreate;
675 struct g_gate_ctl_cancel ggiocancel;
676
677 /*
678 * We communicate with ggate via /dev/ggctl. Open it.
679 */
680 res->hr_ggatefd = open("/dev/" G_GATE_CTL_NAME, O_RDWR);
681 if (res->hr_ggatefd < 0)
682 primary_exit(EX_OSFILE, "Unable to open /dev/" G_GATE_CTL_NAME);
683 /*
684 * Create provider before trying to connect, as connection failure
685 * is not critical, but may take some time.
686 */
687 ggiocreate.gctl_version = G_GATE_VERSION;
688 ggiocreate.gctl_mediasize = res->hr_datasize;
689 ggiocreate.gctl_sectorsize = res->hr_local_sectorsize;
690 ggiocreate.gctl_flags = 0;
691 ggiocreate.gctl_maxcount = G_GATE_MAX_QUEUE_SIZE;
692 ggiocreate.gctl_timeout = 0;
693 ggiocreate.gctl_unit = G_GATE_NAME_GIVEN;
694 snprintf(ggiocreate.gctl_name, sizeof(ggiocreate.gctl_name), "hast/%s",
695 res->hr_provname);
696 bzero(ggiocreate.gctl_info, sizeof(ggiocreate.gctl_info));
697 if (ioctl(res->hr_ggatefd, G_GATE_CMD_CREATE, &ggiocreate) == 0) {
698 pjdlog_info("Device hast/%s created.", res->hr_provname);
699 res->hr_ggateunit = ggiocreate.gctl_unit;
700 return;
701 }
702 if (errno != EEXIST) {
703 primary_exit(EX_OSERR, "Unable to create hast/%s device",
704 res->hr_provname);
705 }
706 pjdlog_debug(1,
707 "Device hast/%s already exists, we will try to take it over.",
708 res->hr_provname);
709 /*
710 * If we received EEXIST, we assume that the process who created the
711 * provider died and didn't clean up. In that case we will start from
712 * where he left of.
713 */
714 ggiocancel.gctl_version = G_GATE_VERSION;
715 ggiocancel.gctl_unit = G_GATE_NAME_GIVEN;
716 snprintf(ggiocancel.gctl_name, sizeof(ggiocancel.gctl_name), "hast/%s",
717 res->hr_provname);
718 if (ioctl(res->hr_ggatefd, G_GATE_CMD_CANCEL, &ggiocancel) == 0) {
719 pjdlog_info("Device hast/%s recovered.", res->hr_provname);
720 res->hr_ggateunit = ggiocancel.gctl_unit;
721 return;
722 }
723 primary_exit(EX_OSERR, "Unable to take over hast/%s device",
724 res->hr_provname);
725}
726
727void
728hastd_primary(struct hast_resource *res)
729{
730 pthread_t td;
731 pid_t pid;
732 int error;
733
734 gres = res;
735
736 /*
737 * Create communication channel between parent and child.
738 */
739 if (proto_client("socketpair://", &res->hr_ctrl) < 0) {
740 KEEP_ERRNO((void)pidfile_remove(pfh));
741 primary_exit(EX_OSERR,
742 "Unable to create control sockets between parent and child");
743 }
744
745 pid = fork();
746 if (pid < 0) {
747 KEEP_ERRNO((void)pidfile_remove(pfh));
748 primary_exit(EX_TEMPFAIL, "Unable to fork");
749 }
750
751 if (pid > 0) {
752 /* This is parent. */
753 res->hr_workerpid = pid;
754 return;
755 }
756 (void)pidfile_close(pfh);
757
758 setproctitle("%s (primary)", res->hr_name);
759
760 init_local(res);
761 if (init_remote(res, NULL, NULL))
762 sync_start();
763 init_ggate(res);
764 init_environment(res);
765 error = pthread_create(&td, NULL, ggate_recv_thread, res);
766 assert(error == 0);
767 error = pthread_create(&td, NULL, local_send_thread, res);
768 assert(error == 0);
769 error = pthread_create(&td, NULL, remote_send_thread, res);
770 assert(error == 0);
771 error = pthread_create(&td, NULL, remote_recv_thread, res);
772 assert(error == 0);
773 error = pthread_create(&td, NULL, ggate_send_thread, res);
774 assert(error == 0);
775 error = pthread_create(&td, NULL, sync_thread, res);
776 assert(error == 0);
777 error = pthread_create(&td, NULL, ctrl_thread, res);
778 assert(error == 0);
779 (void)guard_thread(res);
780}
781
782static void
783reqlog(int loglevel, int debuglevel, struct g_gate_ctl_io *ggio, const char *fmt, ...)
784{
785 char msg[1024];
786 va_list ap;
787 int len;
788
789 va_start(ap, fmt);
790 len = vsnprintf(msg, sizeof(msg), fmt, ap);
791 va_end(ap);
792 if ((size_t)len < sizeof(msg)) {
793 switch (ggio->gctl_cmd) {
794 case BIO_READ:
795 (void)snprintf(msg + len, sizeof(msg) - len,
796 "READ(%ju, %ju).", (uintmax_t)ggio->gctl_offset,
797 (uintmax_t)ggio->gctl_length);
798 break;
799 case BIO_DELETE:
800 (void)snprintf(msg + len, sizeof(msg) - len,
801 "DELETE(%ju, %ju).", (uintmax_t)ggio->gctl_offset,
802 (uintmax_t)ggio->gctl_length);
803 break;
804 case BIO_FLUSH:
805 (void)snprintf(msg + len, sizeof(msg) - len, "FLUSH.");
806 break;
807 case BIO_WRITE:
808 (void)snprintf(msg + len, sizeof(msg) - len,
809 "WRITE(%ju, %ju).", (uintmax_t)ggio->gctl_offset,
810 (uintmax_t)ggio->gctl_length);
811 break;
812 default:
813 (void)snprintf(msg + len, sizeof(msg) - len,
814 "UNKNOWN(%u).", (unsigned int)ggio->gctl_cmd);
815 break;
816 }
817 }
818 pjdlog_common(loglevel, debuglevel, -1, "%s", msg);
819}
820
821static void
822remote_close(struct hast_resource *res, int ncomp)
823{
824
825 rw_wlock(&hio_remote_lock[ncomp]);
826 /*
827 * A race is possible between dropping rlock and acquiring wlock -
828 * another thread can close connection in-between.
829 */
830 if (!ISCONNECTED(res, ncomp)) {
831 assert(res->hr_remotein == NULL);
832 assert(res->hr_remoteout == NULL);
833 rw_unlock(&hio_remote_lock[ncomp]);
834 return;
835 }
836
837 assert(res->hr_remotein != NULL);
838 assert(res->hr_remoteout != NULL);
839
840 pjdlog_debug(2, "Closing old incoming connection to %s.",
841 res->hr_remoteaddr);
842 proto_close(res->hr_remotein);
843 res->hr_remotein = NULL;
844 pjdlog_debug(2, "Closing old outgoing connection to %s.",
845 res->hr_remoteaddr);
846 proto_close(res->hr_remoteout);
847 res->hr_remoteout = NULL;
848
849 rw_unlock(&hio_remote_lock[ncomp]);
850
851 /*
852 * Stop synchronization if in-progress.
853 */
854 mtx_lock(&sync_lock);
855 if (sync_inprogress)
856 sync_inprogress = false;
857 mtx_unlock(&sync_lock);
858
859 /*
860 * Wake up guard thread, so it can immediately start reconnect.
861 */
862 mtx_lock(&hio_guard_lock);
863 cv_signal(&hio_guard_cond);
864 mtx_unlock(&hio_guard_lock);
865}
866
867/*
868 * Thread receives ggate I/O requests from the kernel and passes them to
869 * appropriate threads:
870 * WRITE - always goes to both local_send and remote_send threads
871 * READ (when the block is up-to-date on local component) -
872 * only local_send thread
873 * READ (when the block isn't up-to-date on local component) -
874 * only remote_send thread
875 * DELETE - always goes to both local_send and remote_send threads
876 * FLUSH - always goes to both local_send and remote_send threads
877 */
878static void *
879ggate_recv_thread(void *arg)
880{
881 struct hast_resource *res = arg;
882 struct g_gate_ctl_io *ggio;
883 struct hio *hio;
884 unsigned int ii, ncomp, ncomps;
885 int error;
886
887 ncomps = HAST_NCOMPONENTS;
888
889 for (;;) {
890 pjdlog_debug(2, "ggate_recv: Taking free request.");
891 QUEUE_TAKE2(hio, free);
892 pjdlog_debug(2, "ggate_recv: (%p) Got free request.", hio);
893 ggio = &hio->hio_ggio;
894 ggio->gctl_unit = res->hr_ggateunit;
895 ggio->gctl_length = MAXPHYS;
896 ggio->gctl_error = 0;
897 pjdlog_debug(2,
898 "ggate_recv: (%p) Waiting for request from the kernel.",
899 hio);
900 if (ioctl(res->hr_ggatefd, G_GATE_CMD_START, ggio) < 0) {
901 if (sigexit_received)
902 pthread_exit(NULL);
903 primary_exit(EX_OSERR, "G_GATE_CMD_START failed");
904 }
905 error = ggio->gctl_error;
906 switch (error) {
907 case 0:
908 break;
909 case ECANCELED:
910 /* Exit gracefully. */
911 if (!sigexit_received) {
912 pjdlog_debug(2,
913 "ggate_recv: (%p) Received cancel from the kernel.",
914 hio);
915 pjdlog_info("Received cancel from the kernel, exiting.");
916 }
917 pthread_exit(NULL);
918 case ENOMEM:
919 /*
920 * Buffer too small? Impossible, we allocate MAXPHYS
921 * bytes - request can't be bigger than that.
922 */
923 /* FALLTHROUGH */
924 case ENXIO:
925 default:
926 primary_exitx(EX_OSERR, "G_GATE_CMD_START failed: %s.",
927 strerror(error));
928 }
929 for (ii = 0; ii < ncomps; ii++)
930 hio->hio_errors[ii] = EINVAL;
931 reqlog(LOG_DEBUG, 2, ggio,
932 "ggate_recv: (%p) Request received from the kernel: ",
933 hio);
934 /*
935 * Inform all components about new write request.
936 * For read request prefer local component unless the given
937 * range is out-of-date, then use remote component.
938 */
939 switch (ggio->gctl_cmd) {
940 case BIO_READ:
941 pjdlog_debug(2,
942 "ggate_recv: (%p) Moving request to the send queue.",
943 hio);
944 refcount_init(&hio->hio_countdown, 1);
945 mtx_lock(&metadata_lock);
946 if (res->hr_syncsrc == HAST_SYNCSRC_UNDEF ||
947 res->hr_syncsrc == HAST_SYNCSRC_PRIMARY) {
948 /*
949 * This range is up-to-date on local component,
950 * so handle request locally.
951 */
952 /* Local component is 0 for now. */
953 ncomp = 0;
954 } else /* if (res->hr_syncsrc ==
955 HAST_SYNCSRC_SECONDARY) */ {
956 assert(res->hr_syncsrc ==
957 HAST_SYNCSRC_SECONDARY);
958 /*
959 * This range is out-of-date on local component,
960 * so send request to the remote node.
961 */
962 /* Remote component is 1 for now. */
963 ncomp = 1;
964 }
965 mtx_unlock(&metadata_lock);
966 QUEUE_INSERT1(hio, send, ncomp);
967 break;
968 case BIO_WRITE:
969 for (;;) {
970 mtx_lock(&range_lock);
971 if (rangelock_islocked(range_sync,
972 ggio->gctl_offset, ggio->gctl_length)) {
973 pjdlog_debug(2,
974 "regular: Range offset=%jd length=%zu locked.",
975 (intmax_t)ggio->gctl_offset,
976 (size_t)ggio->gctl_length);
977 range_regular_wait = true;
978 cv_wait(&range_regular_cond, &range_lock);
979 range_regular_wait = false;
980 mtx_unlock(&range_lock);
981 continue;
982 }
983 if (rangelock_add(range_regular,
984 ggio->gctl_offset, ggio->gctl_length) < 0) {
985 mtx_unlock(&range_lock);
986 pjdlog_debug(2,
987 "regular: Range offset=%jd length=%zu is already locked, waiting.",
988 (intmax_t)ggio->gctl_offset,
989 (size_t)ggio->gctl_length);
990 sleep(1);
991 continue;
992 }
993 mtx_unlock(&range_lock);
994 break;
995 }
996 mtx_lock(&res->hr_amp_lock);
997 if (activemap_write_start(res->hr_amp,
998 ggio->gctl_offset, ggio->gctl_length)) {
999 (void)hast_activemap_flush(res);
1000 }
1001 mtx_unlock(&res->hr_amp_lock);
1002 /* FALLTHROUGH */
1003 case BIO_DELETE:
1004 case BIO_FLUSH:
1005 pjdlog_debug(2,
1006 "ggate_recv: (%p) Moving request to the send queues.",
1007 hio);
1008 refcount_init(&hio->hio_countdown, ncomps);
1009 for (ii = 0; ii < ncomps; ii++)
1010 QUEUE_INSERT1(hio, send, ii);
1011 break;
1012 }
1013 }
1014 /* NOTREACHED */
1015 return (NULL);
1016}
1017
1018/*
1019 * Thread reads from or writes to local component.
1020 * If local read fails, it redirects it to remote_send thread.
1021 */
1022static void *
1023local_send_thread(void *arg)
1024{
1025 struct hast_resource *res = arg;
1026 struct g_gate_ctl_io *ggio;
1027 struct hio *hio;
1028 unsigned int ncomp, rncomp;
1029 ssize_t ret;
1030
1031 /* Local component is 0 for now. */
1032 ncomp = 0;
1033 /* Remote component is 1 for now. */
1034 rncomp = 1;
1035
1036 for (;;) {
1037 pjdlog_debug(2, "local_send: Taking request.");
1038 QUEUE_TAKE1(hio, send, ncomp);
1039 pjdlog_debug(2, "local_send: (%p) Got request.", hio);
1040 ggio = &hio->hio_ggio;
1041 switch (ggio->gctl_cmd) {
1042 case BIO_READ:
1043 ret = pread(res->hr_localfd, ggio->gctl_data,
1044 ggio->gctl_length,
1045 ggio->gctl_offset + res->hr_localoff);
1046 if (ret == ggio->gctl_length)
1047 hio->hio_errors[ncomp] = 0;
1048 else {
1049 /*
1050 * If READ failed, try to read from remote node.
1051 */
1052 QUEUE_INSERT1(hio, send, rncomp);
1053 continue;
1054 }
1055 break;
1056 case BIO_WRITE:
1057 ret = pwrite(res->hr_localfd, ggio->gctl_data,
1058 ggio->gctl_length,
1059 ggio->gctl_offset + res->hr_localoff);
1060 if (ret < 0)
1061 hio->hio_errors[ncomp] = errno;
1062 else if (ret != ggio->gctl_length)
1063 hio->hio_errors[ncomp] = EIO;
1064 else
1065 hio->hio_errors[ncomp] = 0;
1066 break;
1067 case BIO_DELETE:
1068 ret = g_delete(res->hr_localfd,
1069 ggio->gctl_offset + res->hr_localoff,
1070 ggio->gctl_length);
1071 if (ret < 0)
1072 hio->hio_errors[ncomp] = errno;
1073 else
1074 hio->hio_errors[ncomp] = 0;
1075 break;
1076 case BIO_FLUSH:
1077 ret = g_flush(res->hr_localfd);
1078 if (ret < 0)
1079 hio->hio_errors[ncomp] = errno;
1080 else
1081 hio->hio_errors[ncomp] = 0;
1082 break;
1083 }
1084 if (refcount_release(&hio->hio_countdown)) {
1085 if (ISSYNCREQ(hio)) {
1086 mtx_lock(&sync_lock);
1087 SYNCREQDONE(hio);
1088 mtx_unlock(&sync_lock);
1089 cv_signal(&sync_cond);
1090 } else {
1091 pjdlog_debug(2,
1092 "local_send: (%p) Moving request to the done queue.",
1093 hio);
1094 QUEUE_INSERT2(hio, done);
1095 }
1096 }
1097 }
1098 /* NOTREACHED */
1099 return (NULL);
1100}
1101
1102/*
1103 * Thread sends request to secondary node.
1104 */
1105static void *
1106remote_send_thread(void *arg)
1107{
1108 struct hast_resource *res = arg;
1109 struct g_gate_ctl_io *ggio;
1110 struct hio *hio;
1111 struct nv *nv;
1112 unsigned int ncomp;
1113 bool wakeup;
1114 uint64_t offset, length;
1115 uint8_t cmd;
1116 void *data;
1117
1118 /* Remote component is 1 for now. */
1119 ncomp = 1;
1120
1121 for (;;) {
1122 pjdlog_debug(2, "remote_send: Taking request.");
1123 QUEUE_TAKE1(hio, send, ncomp);
1124 pjdlog_debug(2, "remote_send: (%p) Got request.", hio);
1125 ggio = &hio->hio_ggio;
1126 switch (ggio->gctl_cmd) {
1127 case BIO_READ:
1128 cmd = HIO_READ;
1129 data = NULL;
1130 offset = ggio->gctl_offset;
1131 length = ggio->gctl_length;
1132 break;
1133 case BIO_WRITE:
1134 cmd = HIO_WRITE;
1135 data = ggio->gctl_data;
1136 offset = ggio->gctl_offset;
1137 length = ggio->gctl_length;
1138 break;
1139 case BIO_DELETE:
1140 cmd = HIO_DELETE;
1141 data = NULL;
1142 offset = ggio->gctl_offset;
1143 length = ggio->gctl_length;
1144 break;
1145 case BIO_FLUSH:
1146 cmd = HIO_FLUSH;
1147 data = NULL;
1148 offset = 0;
1149 length = 0;
1150 break;
1151 default:
1152 assert(!"invalid condition");
1153 abort();
1154 }
1155 nv = nv_alloc();
1156 nv_add_uint8(nv, cmd, "cmd");
1157 nv_add_uint64(nv, (uint64_t)ggio->gctl_seq, "seq");
1158 nv_add_uint64(nv, offset, "offset");
1159 nv_add_uint64(nv, length, "length");
1160 if (nv_error(nv) != 0) {
1161 hio->hio_errors[ncomp] = nv_error(nv);
1162 pjdlog_debug(2,
1163 "remote_send: (%p) Unable to prepare header to send.",
1164 hio);
1165 reqlog(LOG_ERR, 0, ggio,
1166 "Unable to prepare header to send (%s): ",
1167 strerror(nv_error(nv)));
1168 /* Move failed request immediately to the done queue. */
1169 goto done_queue;
1170 }
1171 pjdlog_debug(2,
1172 "remote_send: (%p) Moving request to the recv queue.",
1173 hio);
1174 /*
1175 * Protect connection from disappearing.
1176 */
1177 rw_rlock(&hio_remote_lock[ncomp]);
1178 if (!ISCONNECTED(res, ncomp)) {
1179 rw_unlock(&hio_remote_lock[ncomp]);
1180 hio->hio_errors[ncomp] = ENOTCONN;
1181 goto done_queue;
1182 }
1183 /*
1184 * Move the request to recv queue before sending it, because
1185 * in different order we can get reply before we move request
1186 * to recv queue.
1187 */
1188 mtx_lock(&hio_recv_list_lock[ncomp]);
1189 wakeup = TAILQ_EMPTY(&hio_recv_list[ncomp]);
1190 TAILQ_INSERT_TAIL(&hio_recv_list[ncomp], hio, hio_next[ncomp]);
1191 mtx_unlock(&hio_recv_list_lock[ncomp]);
1192 if (hast_proto_send(res, res->hr_remoteout, nv, data,
1193 data != NULL ? length : 0) < 0) {
1194 hio->hio_errors[ncomp] = errno;
1195 rw_unlock(&hio_remote_lock[ncomp]);
1196 remote_close(res, ncomp);
1197 pjdlog_debug(2,
1198 "remote_send: (%p) Unable to send request.", hio);
1199 reqlog(LOG_ERR, 0, ggio,
1200 "Unable to send request (%s): ",
1201 strerror(hio->hio_errors[ncomp]));
1202 /*
1203 * Take request back from the receive queue and move
1204 * it immediately to the done queue.
1205 */
1206 mtx_lock(&hio_recv_list_lock[ncomp]);
1207 TAILQ_REMOVE(&hio_recv_list[ncomp], hio, hio_next[ncomp]);
1208 mtx_unlock(&hio_recv_list_lock[ncomp]);
1209 goto done_queue;
1210 }
1211 rw_unlock(&hio_remote_lock[ncomp]);
1212 nv_free(nv);
1213 if (wakeup)
1214 cv_signal(&hio_recv_list_cond[ncomp]);
1215 continue;
1216done_queue:
1217 nv_free(nv);
1218 if (ISSYNCREQ(hio)) {
1219 if (!refcount_release(&hio->hio_countdown))
1220 continue;
1221 mtx_lock(&sync_lock);
1222 SYNCREQDONE(hio);
1223 mtx_unlock(&sync_lock);
1224 cv_signal(&sync_cond);
1225 continue;
1226 }
1227 if (ggio->gctl_cmd == BIO_WRITE) {
1228 mtx_lock(&res->hr_amp_lock);
1229 if (activemap_need_sync(res->hr_amp, ggio->gctl_offset,
1230 ggio->gctl_length)) {
1231 (void)hast_activemap_flush(res);
1232 }
1233 mtx_unlock(&res->hr_amp_lock);
1234 }
1235 if (!refcount_release(&hio->hio_countdown))
1236 continue;
1237 pjdlog_debug(2,
1238 "remote_send: (%p) Moving request to the done queue.",
1239 hio);
1240 QUEUE_INSERT2(hio, done);
1241 }
1242 /* NOTREACHED */
1243 return (NULL);
1244}
1245
1246/*
1247 * Thread receives answer from secondary node and passes it to ggate_send
1248 * thread.
1249 */
1250static void *
1251remote_recv_thread(void *arg)
1252{
1253 struct hast_resource *res = arg;
1254 struct g_gate_ctl_io *ggio;
1255 struct hio *hio;
1256 struct nv *nv;
1257 unsigned int ncomp;
1258 uint64_t seq;
1259 int error;
1260
1261 /* Remote component is 1 for now. */
1262 ncomp = 1;
1263
1264 for (;;) {
1265 /* Wait until there is anything to receive. */
1266 mtx_lock(&hio_recv_list_lock[ncomp]);
1267 while (TAILQ_EMPTY(&hio_recv_list[ncomp])) {
1268 pjdlog_debug(2, "remote_recv: No requests, waiting.");
1269 cv_wait(&hio_recv_list_cond[ncomp],
1270 &hio_recv_list_lock[ncomp]);
1271 }
1272 mtx_unlock(&hio_recv_list_lock[ncomp]);
1273 rw_rlock(&hio_remote_lock[ncomp]);
1274 if (!ISCONNECTED(res, ncomp)) {
1275 rw_unlock(&hio_remote_lock[ncomp]);
1276 /*
1277 * Connection is dead, so move all pending requests to
1278 * the done queue (one-by-one).
1279 */
1280 mtx_lock(&hio_recv_list_lock[ncomp]);
1281 hio = TAILQ_FIRST(&hio_recv_list[ncomp]);
1282 assert(hio != NULL);
1283 TAILQ_REMOVE(&hio_recv_list[ncomp], hio,
1284 hio_next[ncomp]);
1285 mtx_unlock(&hio_recv_list_lock[ncomp]);
1286 goto done_queue;
1287 }
1288 if (hast_proto_recv_hdr(res->hr_remotein, &nv) < 0) {
1289 pjdlog_errno(LOG_ERR,
1290 "Unable to receive reply header");
1291 rw_unlock(&hio_remote_lock[ncomp]);
1292 remote_close(res, ncomp);
1293 continue;
1294 }
1295 rw_unlock(&hio_remote_lock[ncomp]);
1296 seq = nv_get_uint64(nv, "seq");
1297 if (seq == 0) {
1298 pjdlog_error("Header contains no 'seq' field.");
1299 nv_free(nv);
1300 continue;
1301 }
1302 mtx_lock(&hio_recv_list_lock[ncomp]);
1303 TAILQ_FOREACH(hio, &hio_recv_list[ncomp], hio_next[ncomp]) {
1304 if (hio->hio_ggio.gctl_seq == seq) {
1305 TAILQ_REMOVE(&hio_recv_list[ncomp], hio,
1306 hio_next[ncomp]);
1307 break;
1308 }
1309 }
1310 mtx_unlock(&hio_recv_list_lock[ncomp]);
1311 if (hio == NULL) {
1312 pjdlog_error("Found no request matching received 'seq' field (%ju).",
1313 (uintmax_t)seq);
1314 nv_free(nv);
1315 continue;
1316 }
1317 error = nv_get_int16(nv, "error");
1318 if (error != 0) {
1319 /* Request failed on remote side. */
1320 hio->hio_errors[ncomp] = 0;
1321 nv_free(nv);
1322 goto done_queue;
1323 }
1324 ggio = &hio->hio_ggio;
1325 switch (ggio->gctl_cmd) {
1326 case BIO_READ:
1327 rw_rlock(&hio_remote_lock[ncomp]);
1328 if (!ISCONNECTED(res, ncomp)) {
1329 rw_unlock(&hio_remote_lock[ncomp]);
1330 nv_free(nv);
1331 goto done_queue;
1332 }
1333 if (hast_proto_recv_data(res, res->hr_remotein, nv,
1334 ggio->gctl_data, ggio->gctl_length) < 0) {
1335 hio->hio_errors[ncomp] = errno;
1336 pjdlog_errno(LOG_ERR,
1337 "Unable to receive reply data");
1338 rw_unlock(&hio_remote_lock[ncomp]);
1339 nv_free(nv);
1340 remote_close(res, ncomp);
1341 goto done_queue;
1342 }
1343 rw_unlock(&hio_remote_lock[ncomp]);
1344 break;
1345 case BIO_WRITE:
1346 case BIO_DELETE:
1347 case BIO_FLUSH:
1348 break;
1349 default:
1350 assert(!"invalid condition");
1351 abort();
1352 }
1353 hio->hio_errors[ncomp] = 0;
1354 nv_free(nv);
1355done_queue:
1356 if (refcount_release(&hio->hio_countdown)) {
1357 if (ISSYNCREQ(hio)) {
1358 mtx_lock(&sync_lock);
1359 SYNCREQDONE(hio);
1360 mtx_unlock(&sync_lock);
1361 cv_signal(&sync_cond);
1362 } else {
1363 pjdlog_debug(2,
1364 "remote_recv: (%p) Moving request to the done queue.",
1365 hio);
1366 QUEUE_INSERT2(hio, done);
1367 }
1368 }
1369 }
1370 /* NOTREACHED */
1371 return (NULL);
1372}
1373
1374/*
1375 * Thread sends answer to the kernel.
1376 */
1377static void *
1378ggate_send_thread(void *arg)
1379{
1380 struct hast_resource *res = arg;
1381 struct g_gate_ctl_io *ggio;
1382 struct hio *hio;
1383 unsigned int ii, ncomp, ncomps;
1384
1385 ncomps = HAST_NCOMPONENTS;
1386
1387 for (;;) {
1388 pjdlog_debug(2, "ggate_send: Taking request.");
1389 QUEUE_TAKE2(hio, done);
1390 pjdlog_debug(2, "ggate_send: (%p) Got request.", hio);
1391 ggio = &hio->hio_ggio;
1392 for (ii = 0; ii < ncomps; ii++) {
1393 if (hio->hio_errors[ii] == 0) {
1394 /*
1395 * One successful request is enough to declare
1396 * success.
1397 */
1398 ggio->gctl_error = 0;
1399 break;
1400 }
1401 }
1402 if (ii == ncomps) {
1403 /*
1404 * None of the requests were successful.
1405 * Use first error.
1406 */
1407 ggio->gctl_error = hio->hio_errors[0];
1408 }
1409 if (ggio->gctl_error == 0 && ggio->gctl_cmd == BIO_WRITE) {
1410 mtx_lock(&res->hr_amp_lock);
1411 activemap_write_complete(res->hr_amp,
1412 ggio->gctl_offset, ggio->gctl_length);
1413 mtx_unlock(&res->hr_amp_lock);
1414 }
1415 if (ggio->gctl_cmd == BIO_WRITE) {
1416 /*
1417 * Unlock range we locked.
1418 */
1419 mtx_lock(&range_lock);
1420 rangelock_del(range_regular, ggio->gctl_offset,
1421 ggio->gctl_length);
1422 if (range_sync_wait)
1423 cv_signal(&range_sync_cond);
1424 mtx_unlock(&range_lock);
1425 /*
1426 * Bump local count if this is first write after
1427 * connection failure with remote node.
1428 */
1429 ncomp = 1;
1430 rw_rlock(&hio_remote_lock[ncomp]);
1431 if (!ISCONNECTED(res, ncomp)) {
1432 mtx_lock(&metadata_lock);
1433 if (res->hr_primary_localcnt ==
1434 res->hr_secondary_remotecnt) {
1435 res->hr_primary_localcnt++;
1436 pjdlog_debug(1,
1437 "Increasing localcnt to %ju.",
1438 (uintmax_t)res->hr_primary_localcnt);
1439 (void)metadata_write(res);
1440 }
1441 mtx_unlock(&metadata_lock);
1442 }
1443 rw_unlock(&hio_remote_lock[ncomp]);
1444 }
1445 if (ioctl(res->hr_ggatefd, G_GATE_CMD_DONE, ggio) < 0)
1446 primary_exit(EX_OSERR, "G_GATE_CMD_DONE failed");
1447 pjdlog_debug(2,
1448 "ggate_send: (%p) Moving request to the free queue.", hio);
1449 QUEUE_INSERT2(hio, free);
1450 }
1451 /* NOTREACHED */
1452 return (NULL);
1453}
1454
1455/*
1456 * Thread synchronize local and remote components.
1457 */
1458static void *
1459sync_thread(void *arg __unused)
1460{
1461 struct hast_resource *res = arg;
1462 struct hio *hio;
1463 struct g_gate_ctl_io *ggio;
1464 unsigned int ii, ncomp, ncomps;
1465 off_t offset, length, synced;
1466 bool dorewind;
1467 int syncext;
1468
1469 ncomps = HAST_NCOMPONENTS;
1470 dorewind = true;
1471 synced = 0;
1472
1473 for (;;) {
1474 mtx_lock(&sync_lock);
1475 while (!sync_inprogress) {
1476 dorewind = true;
1477 synced = 0;
1478 cv_wait(&sync_cond, &sync_lock);
1479 }
1480 mtx_unlock(&sync_lock);
1481 /*
1482 * Obtain offset at which we should synchronize.
1483 * Rewind synchronization if needed.
1484 */
1485 mtx_lock(&res->hr_amp_lock);
1486 if (dorewind)
1487 activemap_sync_rewind(res->hr_amp);
1488 offset = activemap_sync_offset(res->hr_amp, &length, &syncext);
1489 if (syncext != -1) {
1490 /*
1491 * We synchronized entire syncext extent, we can mark
1492 * it as clean now.
1493 */
1494 if (activemap_extent_complete(res->hr_amp, syncext))
1495 (void)hast_activemap_flush(res);
1496 }
1497 mtx_unlock(&res->hr_amp_lock);
1498 if (dorewind) {
1499 dorewind = false;
1500 if (offset < 0)
1501 pjdlog_info("Nodes are in sync.");
1502 else {
1503 pjdlog_info("Synchronization started. %ju bytes to go.",
1504 (uintmax_t)(res->hr_extentsize *
1505 activemap_ndirty(res->hr_amp)));
1506 }
1507 }
1508 if (offset < 0) {
1509 mtx_lock(&sync_lock);
1510 sync_inprogress = false;
1511 mtx_unlock(&sync_lock);
1512 pjdlog_debug(1, "Nothing to synchronize.");
1513 /*
1514 * Synchronization complete, make both localcnt and
1515 * remotecnt equal.
1516 */
1517 ncomp = 1;
1518 rw_rlock(&hio_remote_lock[ncomp]);
1519 if (ISCONNECTED(res, ncomp)) {
1520 if (synced > 0) {
1521 pjdlog_info("Synchronization complete. "
1522 "%jd bytes synchronized.",
1523 (intmax_t)synced);
1524 }
1525 mtx_lock(&metadata_lock);
1526 res->hr_syncsrc = HAST_SYNCSRC_UNDEF;
1527 res->hr_primary_localcnt =
1528 res->hr_secondary_localcnt;
1529 res->hr_primary_remotecnt =
1530 res->hr_secondary_remotecnt;
1531 pjdlog_debug(1,
1532 "Setting localcnt to %ju and remotecnt to %ju.",
1533 (uintmax_t)res->hr_primary_localcnt,
1534 (uintmax_t)res->hr_secondary_localcnt);
1535 (void)metadata_write(res);
1536 mtx_unlock(&metadata_lock);
1537 } else if (synced > 0) {
1538 pjdlog_info("Synchronization interrupted. "
1539 "%jd bytes synchronized so far.",
1540 (intmax_t)synced);
1541 }
1542 rw_unlock(&hio_remote_lock[ncomp]);
1543 continue;
1544 }
1545 pjdlog_debug(2, "sync: Taking free request.");
1546 QUEUE_TAKE2(hio, free);
1547 pjdlog_debug(2, "sync: (%p) Got free request.", hio);
1548 /*
1549 * Lock the range we are going to synchronize. We don't want
1550 * race where someone writes between our read and write.
1551 */
1552 for (;;) {
1553 mtx_lock(&range_lock);
1554 if (rangelock_islocked(range_regular, offset, length)) {
1555 pjdlog_debug(2,
1556 "sync: Range offset=%jd length=%jd locked.",
1557 (intmax_t)offset, (intmax_t)length);
1558 range_sync_wait = true;
1559 cv_wait(&range_sync_cond, &range_lock);
1560 range_sync_wait = false;
1561 mtx_unlock(&range_lock);
1562 continue;
1563 }
1564 if (rangelock_add(range_sync, offset, length) < 0) {
1565 mtx_unlock(&range_lock);
1566 pjdlog_debug(2,
1567 "sync: Range offset=%jd length=%jd is already locked, waiting.",
1568 (intmax_t)offset, (intmax_t)length);
1569 sleep(1);
1570 continue;
1571 }
1572 mtx_unlock(&range_lock);
1573 break;
1574 }
1575 /*
1576 * First read the data from synchronization source.
1577 */
1578 SYNCREQ(hio);
1579 ggio = &hio->hio_ggio;
1580 ggio->gctl_cmd = BIO_READ;
1581 ggio->gctl_offset = offset;
1582 ggio->gctl_length = length;
1583 ggio->gctl_error = 0;
1584 for (ii = 0; ii < ncomps; ii++)
1585 hio->hio_errors[ii] = EINVAL;
1586 reqlog(LOG_DEBUG, 2, ggio, "sync: (%p) Sending sync request: ",
1587 hio);
1588 pjdlog_debug(2, "sync: (%p) Moving request to the send queue.",
1589 hio);
1590 mtx_lock(&metadata_lock);
1591 if (res->hr_syncsrc == HAST_SYNCSRC_PRIMARY) {
1592 /*
1593 * This range is up-to-date on local component,
1594 * so handle request locally.
1595 */
1596 /* Local component is 0 for now. */
1597 ncomp = 0;
1598 } else /* if (res->hr_syncsrc == HAST_SYNCSRC_SECONDARY) */ {
1599 assert(res->hr_syncsrc == HAST_SYNCSRC_SECONDARY);
1600 /*
1601 * This range is out-of-date on local component,
1602 * so send request to the remote node.
1603 */
1604 /* Remote component is 1 for now. */
1605 ncomp = 1;
1606 }
1607 mtx_unlock(&metadata_lock);
1608 refcount_init(&hio->hio_countdown, 1);
1609 QUEUE_INSERT1(hio, send, ncomp);
1610
1611 /*
1612 * Let's wait for READ to finish.
1613 */
1614 mtx_lock(&sync_lock);
1615 while (!ISSYNCREQDONE(hio))
1616 cv_wait(&sync_cond, &sync_lock);
1617 mtx_unlock(&sync_lock);
1618
1619 if (hio->hio_errors[ncomp] != 0) {
1620 pjdlog_error("Unable to read synchronization data: %s.",
1621 strerror(hio->hio_errors[ncomp]));
1622 goto free_queue;
1623 }
1624
1625 /*
1626 * We read the data from synchronization source, now write it
1627 * to synchronization target.
1628 */
1629 SYNCREQ(hio);
1630 ggio->gctl_cmd = BIO_WRITE;
1631 for (ii = 0; ii < ncomps; ii++)
1632 hio->hio_errors[ii] = EINVAL;
1633 reqlog(LOG_DEBUG, 2, ggio, "sync: (%p) Sending sync request: ",
1634 hio);
1635 pjdlog_debug(2, "sync: (%p) Moving request to the send queue.",
1636 hio);
1637 mtx_lock(&metadata_lock);
1638 if (res->hr_syncsrc == HAST_SYNCSRC_PRIMARY) {
1639 /*
1640 * This range is up-to-date on local component,
1641 * so we update remote component.
1642 */
1643 /* Remote component is 1 for now. */
1644 ncomp = 1;
1645 } else /* if (res->hr_syncsrc == HAST_SYNCSRC_SECONDARY) */ {
1646 assert(res->hr_syncsrc == HAST_SYNCSRC_SECONDARY);
1647 /*
1648 * This range is out-of-date on local component,
1649 * so we update it.
1650 */
1651 /* Local component is 0 for now. */
1652 ncomp = 0;
1653 }
1654 mtx_unlock(&metadata_lock);
1655
1656 pjdlog_debug(2, "sync: (%p) Moving request to the send queues.",
1657 hio);
1658 refcount_init(&hio->hio_countdown, 1);
1659 QUEUE_INSERT1(hio, send, ncomp);
1660
1661 /*
1662 * Let's wait for WRITE to finish.
1663 */
1664 mtx_lock(&sync_lock);
1665 while (!ISSYNCREQDONE(hio))
1666 cv_wait(&sync_cond, &sync_lock);
1667 mtx_unlock(&sync_lock);
1668
1669 if (hio->hio_errors[ncomp] != 0) {
1670 pjdlog_error("Unable to write synchronization data: %s.",
1671 strerror(hio->hio_errors[ncomp]));
1672 goto free_queue;
1673 }
1674free_queue:
1675 mtx_lock(&range_lock);
1676 rangelock_del(range_sync, offset, length);
1677 if (range_regular_wait)
1678 cv_signal(&range_regular_cond);
1679 mtx_unlock(&range_lock);
1680
1681 synced += length;
1682
1683 pjdlog_debug(2, "sync: (%p) Moving request to the free queue.",
1684 hio);
1685 QUEUE_INSERT2(hio, free);
1686 }
1687 /* NOTREACHED */
1688 return (NULL);
1689}
1690
1691static void
1692sighandler(int sig)
1693{
1694 bool unlock;
1695
1696 switch (sig) {
1697 case SIGINT:
1698 case SIGTERM:
1699 sigexit_received = true;
1700 break;
1701 default:
1702 assert(!"invalid condition");
1703 }
1704 /*
1705 * XXX: Racy, but if we cannot obtain hio_guard_lock here, we don't
1706 * want to risk deadlock.
1707 */
1708 unlock = mtx_trylock(&hio_guard_lock);
1709 cv_signal(&hio_guard_cond);
1710 if (unlock)
1711 mtx_unlock(&hio_guard_lock);
1712}
1713
1714/*
1715 * Thread guards remote connections and reconnects when needed, handles
1716 * signals, etc.
1717 */
1718static void *
1719guard_thread(void *arg)
1720{
1721 struct hast_resource *res = arg;
1722 struct proto_conn *in, *out;
1723 unsigned int ii, ncomps;
1724 int timeout;
1725
1726 ncomps = HAST_NCOMPONENTS;
1727 /* The is only one remote component for now. */
1728#define ISREMOTE(no) ((no) == 1)
1729
1730 for (;;) {
1731 if (sigexit_received) {
1732 primary_exitx(EX_OK,
1733 "Termination signal received, exiting.");
1734 }
1735 /*
1736 * If all the connection will be fine, we will sleep until
1737 * someone wakes us up.
1738 * If any of the connections will be broken and we won't be
1739 * able to connect, we will sleep only for RECONNECT_SLEEP
1740 * seconds so we can retry soon.
1741 */
1742 timeout = 0;
1743 pjdlog_debug(2, "remote_guard: Checking connections.");
1744 mtx_lock(&hio_guard_lock);
1745 for (ii = 0; ii < ncomps; ii++) {
1746 if (!ISREMOTE(ii))
1747 continue;
1748 rw_rlock(&hio_remote_lock[ii]);
1749 if (ISCONNECTED(res, ii)) {
1750 assert(res->hr_remotein != NULL);
1751 assert(res->hr_remoteout != NULL);
1752 rw_unlock(&hio_remote_lock[ii]);
1753 pjdlog_debug(2,
1754 "remote_guard: Connection to %s is ok.",
1755 res->hr_remoteaddr);
1756 } else {
1757 assert(res->hr_remotein == NULL);
1758 assert(res->hr_remoteout == NULL);
1759 /*
1760 * Upgrade the lock. It doesn't have to be
1761 * atomic as no other thread can change
1762 * connection status from disconnected to
1763 * connected.
1764 */
1765 rw_unlock(&hio_remote_lock[ii]);
1766 pjdlog_debug(2,
1767 "remote_guard: Reconnecting to %s.",
1768 res->hr_remoteaddr);
1769 in = out = NULL;
1770 if (init_remote(res, &in, &out)) {
1771 rw_wlock(&hio_remote_lock[ii]);
1772 assert(res->hr_remotein == NULL);
1773 assert(res->hr_remoteout == NULL);
1774 assert(in != NULL && out != NULL);
1775 res->hr_remotein = in;
1776 res->hr_remoteout = out;
1777 rw_unlock(&hio_remote_lock[ii]);
1778 pjdlog_info("Successfully reconnected to %s.",
1779 res->hr_remoteaddr);
1780 sync_start();
1781 } else {
1782 /* Both connections should be NULL. */
1783 assert(res->hr_remotein == NULL);
1784 assert(res->hr_remoteout == NULL);
1785 assert(in == NULL && out == NULL);
1786 pjdlog_debug(2,
1787 "remote_guard: Reconnect to %s failed.",
1788 res->hr_remoteaddr);
1789 timeout = RECONNECT_SLEEP;
1790 }
1791 }
1792 }
1793 (void)cv_timedwait(&hio_guard_cond, &hio_guard_lock, timeout);
1794 mtx_unlock(&hio_guard_lock);
1795 }
1796#undef ISREMOTE
1797 /* NOTREACHED */
1798 return (NULL);
1799}
|