32
33#include <sys/param.h>
34#include <sys/time.h>
35#include <sys/bio.h>
36#include <sys/disk.h>
37#include <sys/stat.h>
38
39#include <assert.h>
40#include <err.h>
41#include <errno.h>
42#include <fcntl.h>
43#include <libgeom.h>
44#include <pthread.h>
45#include <stdint.h>
46#include <stdio.h>
47#include <string.h>
48#include <sysexits.h>
49#include <unistd.h>
50
51#include <activemap.h>
52#include <nv.h>
53#include <pjdlog.h>
54
55#include "control.h"
56#include "hast.h"
57#include "hast_proto.h"
58#include "hastd.h"
59#include "metadata.h"
60#include "proto.h"
61#include "subr.h"
62#include "synch.h"
63
64struct hio {
65 uint64_t hio_seq;
66 int hio_error;
67 struct nv *hio_nv;
68 void *hio_data;
69 uint8_t hio_cmd;
70 uint64_t hio_offset;
71 uint64_t hio_length;
72 TAILQ_ENTRY(hio) hio_next;
73};
74
75/*
76 * Free list holds unused structures. When free list is empty, we have to wait
77 * until some in-progress requests are freed.
78 */
79static TAILQ_HEAD(, hio) hio_free_list;
80static pthread_mutex_t hio_free_list_lock;
81static pthread_cond_t hio_free_list_cond;
82/*
83 * Disk thread (the one that do I/O requests) takes requests from this list.
84 */
85static TAILQ_HEAD(, hio) hio_disk_list;
86static pthread_mutex_t hio_disk_list_lock;
87static pthread_cond_t hio_disk_list_cond;
88/*
89 * There is one recv list for every component, although local components don't
90 * use recv lists as local requests are done synchronously.
91 */
92static TAILQ_HEAD(, hio) hio_send_list;
93static pthread_mutex_t hio_send_list_lock;
94static pthread_cond_t hio_send_list_cond;
95
96/*
97 * Maximum number of outstanding I/O requests.
98 */
99#define HAST_HIO_MAX 256
100
101static void *recv_thread(void *arg);
102static void *disk_thread(void *arg);
103static void *send_thread(void *arg);
104
105static void
106init_environment(void)
107{
108 struct hio *hio;
109 unsigned int ii;
110
111 /*
112 * Initialize lists, their locks and theirs condition variables.
113 */
114 TAILQ_INIT(&hio_free_list);
115 mtx_init(&hio_free_list_lock);
116 cv_init(&hio_free_list_cond);
117 TAILQ_INIT(&hio_disk_list);
118 mtx_init(&hio_disk_list_lock);
119 cv_init(&hio_disk_list_cond);
120 TAILQ_INIT(&hio_send_list);
121 mtx_init(&hio_send_list_lock);
122 cv_init(&hio_send_list_cond);
123
124 /*
125 * Allocate requests pool and initialize requests.
126 */
127 for (ii = 0; ii < HAST_HIO_MAX; ii++) {
128 hio = malloc(sizeof(*hio));
129 if (hio == NULL) {
130 errx(EX_TEMPFAIL, "cannot allocate %zu bytes of memory "
131 "for hio request", sizeof(*hio));
132 }
133 hio->hio_error = 0;
134 hio->hio_data = malloc(MAXPHYS);
135 if (hio->hio_data == NULL) {
136 errx(EX_TEMPFAIL, "cannot allocate %zu bytes of memory "
137 "for gctl_data", (size_t)MAXPHYS);
138 }
139 TAILQ_INSERT_HEAD(&hio_free_list, hio, hio_next);
140 }
141}
142
143static void
144init_local(struct hast_resource *res)
145{
146
147 if (metadata_read(res, true) < 0)
148 exit(EX_NOINPUT);
149}
150
151static void
152init_remote(struct hast_resource *res, struct nv *nvin)
153{
154 uint64_t resuid;
155 struct nv *nvout;
156 unsigned char *map;
157 size_t mapsize;
158
159 map = NULL;
160 mapsize = 0;
161 nvout = nv_alloc();
162 nv_add_int64(nvout, (int64_t)res->hr_datasize, "datasize");
163 nv_add_int32(nvout, (int32_t)res->hr_extentsize, "extentsize");
164 resuid = nv_get_uint64(nvin, "resuid");
165 res->hr_primary_localcnt = nv_get_uint64(nvin, "localcnt");
166 res->hr_primary_remotecnt = nv_get_uint64(nvin, "remotecnt");
167 nv_add_uint64(nvout, res->hr_secondary_localcnt, "localcnt");
168 nv_add_uint64(nvout, res->hr_secondary_remotecnt, "remotecnt");
169 mapsize = activemap_calc_ondisk_size(res->hr_local_mediasize -
170 METADATA_SIZE, res->hr_extentsize, res->hr_local_sectorsize);
171 map = malloc(mapsize);
172 if (map == NULL) {
173 pjdlog_exitx(EX_TEMPFAIL,
174 "Unable to allocate memory (%zu bytes) for activemap.",
175 mapsize);
176 }
177 nv_add_uint32(nvout, (uint32_t)mapsize, "mapsize");
178 /*
179 * When we work as primary and secondary is missing we will increase
180 * localcnt in our metadata. When secondary is connected and synced
181 * we make localcnt be equal to remotecnt, which means nodes are more
182 * or less in sync.
183 * Split-brain condition is when both nodes are not able to communicate
184 * and are both configured as primary nodes. In turn, they can both
185 * make incompatible changes to the data and we have to detect that.
186 * Under split-brain condition we will increase our localcnt on first
187 * write and remote node will increase its localcnt on first write.
188 * When we connect we can see that primary's localcnt is greater than
189 * our remotecnt (primary was modified while we weren't watching) and
190 * our localcnt is greater than primary's remotecnt (we were modified
191 * while primary wasn't watching).
192 * There are many possible combinations which are all gathered below.
193 * Don't pay too much attention to exact numbers, the more important
194 * is to compare them. We compare secondary's local with primary's
195 * remote and secondary's remote with primary's local.
196 * Note that every case where primary's localcnt is smaller than
197 * secondary's remotecnt and where secondary's localcnt is smaller than
198 * primary's remotecnt should be impossible in practise. We will perform
199 * full synchronization then. Those cases are marked with an asterisk.
200 * Regular synchronization means that only extents marked as dirty are
201 * synchronized (regular synchronization).
202 *
203 * SECONDARY METADATA PRIMARY METADATA
204 * local=3 remote=3 local=2 remote=2* ?! Full sync from secondary.
205 * local=3 remote=3 local=2 remote=3* ?! Full sync from primary.
206 * local=3 remote=3 local=2 remote=4* ?! Full sync from primary.
207 * local=3 remote=3 local=3 remote=2 Primary is out-of-date,
208 * regular sync from secondary.
209 * local=3 remote=3 local=3 remote=3 Regular sync just in case.
210 * local=3 remote=3 local=3 remote=4* ?! Full sync from primary.
211 * local=3 remote=3 local=4 remote=2 Split-brain condition.
212 * local=3 remote=3 local=4 remote=3 Secondary out-of-date,
213 * regular sync from primary.
214 * local=3 remote=3 local=4 remote=4* ?! Full sync from primary.
215 */
216 if (res->hr_resuid == 0) {
217 /*
218 * Provider is used for the first time. Initialize everything.
219 */
220 assert(res->hr_secondary_localcnt == 0);
221 res->hr_resuid = resuid;
222 if (metadata_write(res) < 0)
223 exit(EX_NOINPUT);
224 memset(map, 0xff, mapsize);
225 nv_add_uint8(nvout, HAST_SYNCSRC_PRIMARY, "syncsrc");
226 } else if (
227 /* Is primary is out-of-date? */
228 (res->hr_secondary_localcnt > res->hr_primary_remotecnt &&
229 res->hr_secondary_remotecnt == res->hr_primary_localcnt) ||
230 /* Node are more or less in sync? */
231 (res->hr_secondary_localcnt == res->hr_primary_remotecnt &&
232 res->hr_secondary_remotecnt == res->hr_primary_localcnt) ||
233 /* Is secondary is out-of-date? */
234 (res->hr_secondary_localcnt == res->hr_primary_remotecnt &&
235 res->hr_secondary_remotecnt < res->hr_primary_localcnt)) {
236 /*
237 * Nodes are more or less in sync or one of the nodes is
238 * out-of-date.
239 * It doesn't matter at this point which one, we just have to
240 * send out local bitmap to the remote node.
241 */
242 if (pread(res->hr_localfd, map, mapsize, METADATA_SIZE) !=
243 (ssize_t)mapsize) {
244 pjdlog_exit(LOG_ERR, "Unable to read activemap");
245 }
246 if (res->hr_secondary_localcnt > res->hr_primary_remotecnt &&
247 res->hr_secondary_remotecnt == res->hr_primary_localcnt) {
248 /* Primary is out-of-date, sync from secondary. */
249 nv_add_uint8(nvout, HAST_SYNCSRC_SECONDARY, "syncsrc");
250 } else {
251 /*
252 * Secondary is out-of-date or counts match.
253 * Sync from primary.
254 */
255 nv_add_uint8(nvout, HAST_SYNCSRC_PRIMARY, "syncsrc");
256 }
257 } else if (res->hr_secondary_localcnt > res->hr_primary_remotecnt &&
258 res->hr_primary_localcnt > res->hr_secondary_remotecnt) {
259 /*
260 * Not good, we have split-brain condition.
261 */
262 pjdlog_error("Split-brain detected, exiting.");
263 nv_add_string(nvout, "Split-brain condition!", "errmsg");
264 free(map);
265 map = NULL;
266 mapsize = 0;
267 } else /* if (res->hr_secondary_localcnt < res->hr_primary_remotecnt ||
268 res->hr_primary_localcnt < res->hr_secondary_remotecnt) */ {
269 /*
270 * This should never happen in practise, but we will perform
271 * full synchronization.
272 */
273 assert(res->hr_secondary_localcnt < res->hr_primary_remotecnt ||
274 res->hr_primary_localcnt < res->hr_secondary_remotecnt);
275 mapsize = activemap_calc_ondisk_size(res->hr_local_mediasize -
276 METADATA_SIZE, res->hr_extentsize,
277 res->hr_local_sectorsize);
278 memset(map, 0xff, mapsize);
279 if (res->hr_secondary_localcnt > res->hr_primary_remotecnt) {
280 /* In this one of five cases sync from secondary. */
281 nv_add_uint8(nvout, HAST_SYNCSRC_SECONDARY, "syncsrc");
282 } else {
283 /* For the rest four cases sync from primary. */
284 nv_add_uint8(nvout, HAST_SYNCSRC_PRIMARY, "syncsrc");
285 }
286 pjdlog_warning("This should never happen, asking for full synchronization (primary(local=%ju, remote=%ju), secondary(local=%ju, remote=%ju)).",
287 (uintmax_t)res->hr_primary_localcnt,
288 (uintmax_t)res->hr_primary_remotecnt,
289 (uintmax_t)res->hr_secondary_localcnt,
290 (uintmax_t)res->hr_secondary_remotecnt);
291 }
292 if (hast_proto_send(res, res->hr_remotein, nvout, map, mapsize) < 0) {
293 pjdlog_errno(LOG_WARNING, "Unable to send activemap to %s",
294 res->hr_remoteaddr);
295 nv_free(nvout);
296 exit(EX_TEMPFAIL);
297 }
298 if (res->hr_secondary_localcnt > res->hr_primary_remotecnt &&
299 res->hr_primary_localcnt > res->hr_secondary_remotecnt) {
300 /* Exit on split-brain. */
301 exit(EX_CONFIG);
302 }
303}
304
305void
306hastd_secondary(struct hast_resource *res, struct nv *nvin)
307{
308 pthread_t td;
309 pid_t pid;
310 int error;
311
312 /*
313 * Create communication channel between parent and child.
314 */
315 if (proto_client("socketpair://", &res->hr_ctrl) < 0) {
316 KEEP_ERRNO((void)pidfile_remove(pfh));
317 pjdlog_exit(EX_OSERR,
318 "Unable to create control sockets between parent and child");
319 }
320
321 pid = fork();
322 if (pid < 0) {
323 KEEP_ERRNO((void)pidfile_remove(pfh));
324 pjdlog_exit(EX_OSERR, "Unable to fork");
325 }
326
327 if (pid > 0) {
328 /* This is parent. */
329 proto_close(res->hr_remotein);
330 res->hr_remotein = NULL;
331 proto_close(res->hr_remoteout);
332 res->hr_remoteout = NULL;
333 res->hr_workerpid = pid;
334 return;
335 }
336 (void)pidfile_close(pfh);
337
338 setproctitle("%s (secondary)", res->hr_name);
339
| 32
33#include <sys/param.h>
34#include <sys/time.h>
35#include <sys/bio.h>
36#include <sys/disk.h>
37#include <sys/stat.h>
38
39#include <assert.h>
40#include <err.h>
41#include <errno.h>
42#include <fcntl.h>
43#include <libgeom.h>
44#include <pthread.h>
45#include <stdint.h>
46#include <stdio.h>
47#include <string.h>
48#include <sysexits.h>
49#include <unistd.h>
50
51#include <activemap.h>
52#include <nv.h>
53#include <pjdlog.h>
54
55#include "control.h"
56#include "hast.h"
57#include "hast_proto.h"
58#include "hastd.h"
59#include "metadata.h"
60#include "proto.h"
61#include "subr.h"
62#include "synch.h"
63
64struct hio {
65 uint64_t hio_seq;
66 int hio_error;
67 struct nv *hio_nv;
68 void *hio_data;
69 uint8_t hio_cmd;
70 uint64_t hio_offset;
71 uint64_t hio_length;
72 TAILQ_ENTRY(hio) hio_next;
73};
74
75/*
76 * Free list holds unused structures. When free list is empty, we have to wait
77 * until some in-progress requests are freed.
78 */
79static TAILQ_HEAD(, hio) hio_free_list;
80static pthread_mutex_t hio_free_list_lock;
81static pthread_cond_t hio_free_list_cond;
82/*
83 * Disk thread (the one that do I/O requests) takes requests from this list.
84 */
85static TAILQ_HEAD(, hio) hio_disk_list;
86static pthread_mutex_t hio_disk_list_lock;
87static pthread_cond_t hio_disk_list_cond;
88/*
89 * There is one recv list for every component, although local components don't
90 * use recv lists as local requests are done synchronously.
91 */
92static TAILQ_HEAD(, hio) hio_send_list;
93static pthread_mutex_t hio_send_list_lock;
94static pthread_cond_t hio_send_list_cond;
95
96/*
97 * Maximum number of outstanding I/O requests.
98 */
99#define HAST_HIO_MAX 256
100
101static void *recv_thread(void *arg);
102static void *disk_thread(void *arg);
103static void *send_thread(void *arg);
104
105static void
106init_environment(void)
107{
108 struct hio *hio;
109 unsigned int ii;
110
111 /*
112 * Initialize lists, their locks and theirs condition variables.
113 */
114 TAILQ_INIT(&hio_free_list);
115 mtx_init(&hio_free_list_lock);
116 cv_init(&hio_free_list_cond);
117 TAILQ_INIT(&hio_disk_list);
118 mtx_init(&hio_disk_list_lock);
119 cv_init(&hio_disk_list_cond);
120 TAILQ_INIT(&hio_send_list);
121 mtx_init(&hio_send_list_lock);
122 cv_init(&hio_send_list_cond);
123
124 /*
125 * Allocate requests pool and initialize requests.
126 */
127 for (ii = 0; ii < HAST_HIO_MAX; ii++) {
128 hio = malloc(sizeof(*hio));
129 if (hio == NULL) {
130 errx(EX_TEMPFAIL, "cannot allocate %zu bytes of memory "
131 "for hio request", sizeof(*hio));
132 }
133 hio->hio_error = 0;
134 hio->hio_data = malloc(MAXPHYS);
135 if (hio->hio_data == NULL) {
136 errx(EX_TEMPFAIL, "cannot allocate %zu bytes of memory "
137 "for gctl_data", (size_t)MAXPHYS);
138 }
139 TAILQ_INSERT_HEAD(&hio_free_list, hio, hio_next);
140 }
141}
142
143static void
144init_local(struct hast_resource *res)
145{
146
147 if (metadata_read(res, true) < 0)
148 exit(EX_NOINPUT);
149}
150
151static void
152init_remote(struct hast_resource *res, struct nv *nvin)
153{
154 uint64_t resuid;
155 struct nv *nvout;
156 unsigned char *map;
157 size_t mapsize;
158
159 map = NULL;
160 mapsize = 0;
161 nvout = nv_alloc();
162 nv_add_int64(nvout, (int64_t)res->hr_datasize, "datasize");
163 nv_add_int32(nvout, (int32_t)res->hr_extentsize, "extentsize");
164 resuid = nv_get_uint64(nvin, "resuid");
165 res->hr_primary_localcnt = nv_get_uint64(nvin, "localcnt");
166 res->hr_primary_remotecnt = nv_get_uint64(nvin, "remotecnt");
167 nv_add_uint64(nvout, res->hr_secondary_localcnt, "localcnt");
168 nv_add_uint64(nvout, res->hr_secondary_remotecnt, "remotecnt");
169 mapsize = activemap_calc_ondisk_size(res->hr_local_mediasize -
170 METADATA_SIZE, res->hr_extentsize, res->hr_local_sectorsize);
171 map = malloc(mapsize);
172 if (map == NULL) {
173 pjdlog_exitx(EX_TEMPFAIL,
174 "Unable to allocate memory (%zu bytes) for activemap.",
175 mapsize);
176 }
177 nv_add_uint32(nvout, (uint32_t)mapsize, "mapsize");
178 /*
179 * When we work as primary and secondary is missing we will increase
180 * localcnt in our metadata. When secondary is connected and synced
181 * we make localcnt be equal to remotecnt, which means nodes are more
182 * or less in sync.
183 * Split-brain condition is when both nodes are not able to communicate
184 * and are both configured as primary nodes. In turn, they can both
185 * make incompatible changes to the data and we have to detect that.
186 * Under split-brain condition we will increase our localcnt on first
187 * write and remote node will increase its localcnt on first write.
188 * When we connect we can see that primary's localcnt is greater than
189 * our remotecnt (primary was modified while we weren't watching) and
190 * our localcnt is greater than primary's remotecnt (we were modified
191 * while primary wasn't watching).
192 * There are many possible combinations which are all gathered below.
193 * Don't pay too much attention to exact numbers, the more important
194 * is to compare them. We compare secondary's local with primary's
195 * remote and secondary's remote with primary's local.
196 * Note that every case where primary's localcnt is smaller than
197 * secondary's remotecnt and where secondary's localcnt is smaller than
198 * primary's remotecnt should be impossible in practise. We will perform
199 * full synchronization then. Those cases are marked with an asterisk.
200 * Regular synchronization means that only extents marked as dirty are
201 * synchronized (regular synchronization).
202 *
203 * SECONDARY METADATA PRIMARY METADATA
204 * local=3 remote=3 local=2 remote=2* ?! Full sync from secondary.
205 * local=3 remote=3 local=2 remote=3* ?! Full sync from primary.
206 * local=3 remote=3 local=2 remote=4* ?! Full sync from primary.
207 * local=3 remote=3 local=3 remote=2 Primary is out-of-date,
208 * regular sync from secondary.
209 * local=3 remote=3 local=3 remote=3 Regular sync just in case.
210 * local=3 remote=3 local=3 remote=4* ?! Full sync from primary.
211 * local=3 remote=3 local=4 remote=2 Split-brain condition.
212 * local=3 remote=3 local=4 remote=3 Secondary out-of-date,
213 * regular sync from primary.
214 * local=3 remote=3 local=4 remote=4* ?! Full sync from primary.
215 */
216 if (res->hr_resuid == 0) {
217 /*
218 * Provider is used for the first time. Initialize everything.
219 */
220 assert(res->hr_secondary_localcnt == 0);
221 res->hr_resuid = resuid;
222 if (metadata_write(res) < 0)
223 exit(EX_NOINPUT);
224 memset(map, 0xff, mapsize);
225 nv_add_uint8(nvout, HAST_SYNCSRC_PRIMARY, "syncsrc");
226 } else if (
227 /* Is primary is out-of-date? */
228 (res->hr_secondary_localcnt > res->hr_primary_remotecnt &&
229 res->hr_secondary_remotecnt == res->hr_primary_localcnt) ||
230 /* Node are more or less in sync? */
231 (res->hr_secondary_localcnt == res->hr_primary_remotecnt &&
232 res->hr_secondary_remotecnt == res->hr_primary_localcnt) ||
233 /* Is secondary is out-of-date? */
234 (res->hr_secondary_localcnt == res->hr_primary_remotecnt &&
235 res->hr_secondary_remotecnt < res->hr_primary_localcnt)) {
236 /*
237 * Nodes are more or less in sync or one of the nodes is
238 * out-of-date.
239 * It doesn't matter at this point which one, we just have to
240 * send out local bitmap to the remote node.
241 */
242 if (pread(res->hr_localfd, map, mapsize, METADATA_SIZE) !=
243 (ssize_t)mapsize) {
244 pjdlog_exit(LOG_ERR, "Unable to read activemap");
245 }
246 if (res->hr_secondary_localcnt > res->hr_primary_remotecnt &&
247 res->hr_secondary_remotecnt == res->hr_primary_localcnt) {
248 /* Primary is out-of-date, sync from secondary. */
249 nv_add_uint8(nvout, HAST_SYNCSRC_SECONDARY, "syncsrc");
250 } else {
251 /*
252 * Secondary is out-of-date or counts match.
253 * Sync from primary.
254 */
255 nv_add_uint8(nvout, HAST_SYNCSRC_PRIMARY, "syncsrc");
256 }
257 } else if (res->hr_secondary_localcnt > res->hr_primary_remotecnt &&
258 res->hr_primary_localcnt > res->hr_secondary_remotecnt) {
259 /*
260 * Not good, we have split-brain condition.
261 */
262 pjdlog_error("Split-brain detected, exiting.");
263 nv_add_string(nvout, "Split-brain condition!", "errmsg");
264 free(map);
265 map = NULL;
266 mapsize = 0;
267 } else /* if (res->hr_secondary_localcnt < res->hr_primary_remotecnt ||
268 res->hr_primary_localcnt < res->hr_secondary_remotecnt) */ {
269 /*
270 * This should never happen in practise, but we will perform
271 * full synchronization.
272 */
273 assert(res->hr_secondary_localcnt < res->hr_primary_remotecnt ||
274 res->hr_primary_localcnt < res->hr_secondary_remotecnt);
275 mapsize = activemap_calc_ondisk_size(res->hr_local_mediasize -
276 METADATA_SIZE, res->hr_extentsize,
277 res->hr_local_sectorsize);
278 memset(map, 0xff, mapsize);
279 if (res->hr_secondary_localcnt > res->hr_primary_remotecnt) {
280 /* In this one of five cases sync from secondary. */
281 nv_add_uint8(nvout, HAST_SYNCSRC_SECONDARY, "syncsrc");
282 } else {
283 /* For the rest four cases sync from primary. */
284 nv_add_uint8(nvout, HAST_SYNCSRC_PRIMARY, "syncsrc");
285 }
286 pjdlog_warning("This should never happen, asking for full synchronization (primary(local=%ju, remote=%ju), secondary(local=%ju, remote=%ju)).",
287 (uintmax_t)res->hr_primary_localcnt,
288 (uintmax_t)res->hr_primary_remotecnt,
289 (uintmax_t)res->hr_secondary_localcnt,
290 (uintmax_t)res->hr_secondary_remotecnt);
291 }
292 if (hast_proto_send(res, res->hr_remotein, nvout, map, mapsize) < 0) {
293 pjdlog_errno(LOG_WARNING, "Unable to send activemap to %s",
294 res->hr_remoteaddr);
295 nv_free(nvout);
296 exit(EX_TEMPFAIL);
297 }
298 if (res->hr_secondary_localcnt > res->hr_primary_remotecnt &&
299 res->hr_primary_localcnt > res->hr_secondary_remotecnt) {
300 /* Exit on split-brain. */
301 exit(EX_CONFIG);
302 }
303}
304
305void
306hastd_secondary(struct hast_resource *res, struct nv *nvin)
307{
308 pthread_t td;
309 pid_t pid;
310 int error;
311
312 /*
313 * Create communication channel between parent and child.
314 */
315 if (proto_client("socketpair://", &res->hr_ctrl) < 0) {
316 KEEP_ERRNO((void)pidfile_remove(pfh));
317 pjdlog_exit(EX_OSERR,
318 "Unable to create control sockets between parent and child");
319 }
320
321 pid = fork();
322 if (pid < 0) {
323 KEEP_ERRNO((void)pidfile_remove(pfh));
324 pjdlog_exit(EX_OSERR, "Unable to fork");
325 }
326
327 if (pid > 0) {
328 /* This is parent. */
329 proto_close(res->hr_remotein);
330 res->hr_remotein = NULL;
331 proto_close(res->hr_remoteout);
332 res->hr_remoteout = NULL;
333 res->hr_workerpid = pid;
334 return;
335 }
336 (void)pidfile_close(pfh);
337
338 setproctitle("%s (secondary)", res->hr_name);
339
|
340 init_local(res);
341 init_remote(res, nvin);
342 init_environment();
343
344 error = pthread_create(&td, NULL, recv_thread, res);
345 assert(error == 0);
346 error = pthread_create(&td, NULL, disk_thread, res);
347 assert(error == 0);
348 error = pthread_create(&td, NULL, send_thread, res);
349 assert(error == 0);
350 (void)ctrl_thread(res);
351}
352
353static void
354reqlog(int loglevel, int debuglevel, int error, struct hio *hio, const char *fmt, ...)
355{
356 char msg[1024];
357 va_list ap;
358 int len;
359
360 va_start(ap, fmt);
361 len = vsnprintf(msg, sizeof(msg), fmt, ap);
362 va_end(ap);
363 if ((size_t)len < sizeof(msg)) {
364 switch (hio->hio_cmd) {
365 case HIO_READ:
366 (void)snprintf(msg + len, sizeof(msg) - len,
367 "READ(%ju, %ju).", (uintmax_t)hio->hio_offset,
368 (uintmax_t)hio->hio_length);
369 break;
370 case HIO_DELETE:
371 (void)snprintf(msg + len, sizeof(msg) - len,
372 "DELETE(%ju, %ju).", (uintmax_t)hio->hio_offset,
373 (uintmax_t)hio->hio_length);
374 break;
375 case HIO_FLUSH:
376 (void)snprintf(msg + len, sizeof(msg) - len, "FLUSH.");
377 break;
378 case HIO_WRITE:
379 (void)snprintf(msg + len, sizeof(msg) - len,
380 "WRITE(%ju, %ju).", (uintmax_t)hio->hio_offset,
381 (uintmax_t)hio->hio_length);
382 break;
383 default:
384 (void)snprintf(msg + len, sizeof(msg) - len,
385 "UNKNOWN(%u).", (unsigned int)hio->hio_cmd);
386 break;
387 }
388 }
389 pjdlog_common(loglevel, debuglevel, error, "%s", msg);
390}
391
392static int
393requnpack(struct hast_resource *res, struct hio *hio)
394{
395
396 hio->hio_cmd = nv_get_uint8(hio->hio_nv, "cmd");
397 if (hio->hio_cmd == 0) {
398 pjdlog_error("Header contains no 'cmd' field.");
399 hio->hio_error = EINVAL;
400 goto end;
401 }
402 switch (hio->hio_cmd) {
403 case HIO_READ:
404 case HIO_WRITE:
405 case HIO_DELETE:
406 hio->hio_offset = nv_get_uint64(hio->hio_nv, "offset");
407 if (nv_error(hio->hio_nv) != 0) {
408 pjdlog_error("Header is missing 'offset' field.");
409 hio->hio_error = EINVAL;
410 goto end;
411 }
412 hio->hio_length = nv_get_uint64(hio->hio_nv, "length");
413 if (nv_error(hio->hio_nv) != 0) {
414 pjdlog_error("Header is missing 'length' field.");
415 hio->hio_error = EINVAL;
416 goto end;
417 }
418 if (hio->hio_length == 0) {
419 pjdlog_error("Data length is zero.");
420 hio->hio_error = EINVAL;
421 goto end;
422 }
423 if (hio->hio_length > MAXPHYS) {
424 pjdlog_error("Data length is too large (%ju > %ju).",
425 (uintmax_t)hio->hio_length, (uintmax_t)MAXPHYS);
426 hio->hio_error = EINVAL;
427 goto end;
428 }
429 if ((hio->hio_offset % res->hr_local_sectorsize) != 0) {
430 pjdlog_error("Offset %ju is not multiple of sector size.",
431 (uintmax_t)hio->hio_offset);
432 hio->hio_error = EINVAL;
433 goto end;
434 }
435 if ((hio->hio_length % res->hr_local_sectorsize) != 0) {
436 pjdlog_error("Length %ju is not multiple of sector size.",
437 (uintmax_t)hio->hio_length);
438 hio->hio_error = EINVAL;
439 goto end;
440 }
441 if (hio->hio_offset + hio->hio_length >
442 (uint64_t)res->hr_datasize) {
443 pjdlog_error("Data offset is too large (%ju > %ju).",
444 (uintmax_t)(hio->hio_offset + hio->hio_length),
445 (uintmax_t)res->hr_datasize);
446 hio->hio_error = EINVAL;
447 goto end;
448 }
449 break;
450 default:
451 pjdlog_error("Header contains invalid 'cmd' (%hhu).",
452 hio->hio_cmd);
453 hio->hio_error = EINVAL;
454 goto end;
455 }
456 hio->hio_error = 0;
457end:
458 return (hio->hio_error);
459}
460
461/*
462 * Thread receives requests from the primary node.
463 */
464static void *
465recv_thread(void *arg)
466{
467 struct hast_resource *res = arg;
468 struct hio *hio;
469 bool wakeup;
470
471 for (;;) {
472 pjdlog_debug(2, "recv: Taking free request.");
473 mtx_lock(&hio_free_list_lock);
474 while ((hio = TAILQ_FIRST(&hio_free_list)) == NULL) {
475 pjdlog_debug(2, "recv: No free requests, waiting.");
476 cv_wait(&hio_free_list_cond, &hio_free_list_lock);
477 }
478 TAILQ_REMOVE(&hio_free_list, hio, hio_next);
479 mtx_unlock(&hio_free_list_lock);
480 pjdlog_debug(2, "recv: (%p) Got request.", hio);
481 if (hast_proto_recv_hdr(res->hr_remotein, &hio->hio_nv) < 0) {
482 pjdlog_exit(EX_TEMPFAIL,
483 "Unable to receive request header");
484 }
485 if (requnpack(res, hio) != 0)
486 goto send_queue;
487 reqlog(LOG_DEBUG, 2, -1, hio,
488 "recv: (%p) Got request header: ", hio);
489 if (hio->hio_cmd == HIO_WRITE) {
490 if (hast_proto_recv_data(res, res->hr_remotein,
491 hio->hio_nv, hio->hio_data, MAXPHYS) < 0) {
492 pjdlog_exit(EX_TEMPFAIL,
493 "Unable to receive reply data");
494 }
495 }
496 pjdlog_debug(2, "recv: (%p) Moving request to the disk queue.",
497 hio);
498 mtx_lock(&hio_disk_list_lock);
499 wakeup = TAILQ_EMPTY(&hio_disk_list);
500 TAILQ_INSERT_TAIL(&hio_disk_list, hio, hio_next);
501 mtx_unlock(&hio_disk_list_lock);
502 if (wakeup)
503 cv_signal(&hio_disk_list_cond);
504 continue;
505send_queue:
506 pjdlog_debug(2, "recv: (%p) Moving request to the send queue.",
507 hio);
508 mtx_lock(&hio_send_list_lock);
509 wakeup = TAILQ_EMPTY(&hio_send_list);
510 TAILQ_INSERT_TAIL(&hio_send_list, hio, hio_next);
511 mtx_unlock(&hio_send_list_lock);
512 if (wakeup)
513 cv_signal(&hio_send_list_cond);
514 }
515 /* NOTREACHED */
516 return (NULL);
517}
518
519/*
520 * Thread reads from or writes to local component and also handles DELETE and
521 * FLUSH requests.
522 */
523static void *
524disk_thread(void *arg)
525{
526 struct hast_resource *res = arg;
527 struct hio *hio;
528 ssize_t ret;
529 bool clear_activemap, wakeup;
530
531 clear_activemap = true;
532
533 for (;;) {
534 pjdlog_debug(2, "disk: Taking request.");
535 mtx_lock(&hio_disk_list_lock);
536 while ((hio = TAILQ_FIRST(&hio_disk_list)) == NULL) {
537 pjdlog_debug(2, "disk: No requests, waiting.");
538 cv_wait(&hio_disk_list_cond, &hio_disk_list_lock);
539 }
540 TAILQ_REMOVE(&hio_disk_list, hio, hio_next);
541 mtx_unlock(&hio_disk_list_lock);
542 while (clear_activemap) {
543 unsigned char *map;
544 size_t mapsize;
545
546 /*
547 * When first request is received, it means that primary
548 * already received our activemap, merged it and stored
549 * locally. We can now safely clear our activemap.
550 */
551 mapsize =
552 activemap_calc_ondisk_size(res->hr_local_mediasize -
553 METADATA_SIZE, res->hr_extentsize,
554 res->hr_local_sectorsize);
555 map = calloc(1, mapsize);
556 if (map == NULL) {
557 pjdlog_warning("Unable to allocate memory to clear local activemap.");
558 break;
559 }
560 if (pwrite(res->hr_localfd, map, mapsize,
561 METADATA_SIZE) != (ssize_t)mapsize) {
562 pjdlog_errno(LOG_WARNING,
563 "Unable to store cleared activemap");
564 free(map);
565 break;
566 }
567 free(map);
568 clear_activemap = false;
569 pjdlog_debug(1, "Local activemap cleared.");
570 }
571 reqlog(LOG_DEBUG, 2, -1, hio, "disk: (%p) Got request: ", hio);
572 /* Handle the actual request. */
573 switch (hio->hio_cmd) {
574 case HIO_READ:
575 ret = pread(res->hr_localfd, hio->hio_data,
576 hio->hio_length,
577 hio->hio_offset + res->hr_localoff);
578 if (ret < 0)
579 hio->hio_error = errno;
580 else if (ret != (int64_t)hio->hio_length)
581 hio->hio_error = EIO;
582 else
583 hio->hio_error = 0;
584 break;
585 case HIO_WRITE:
586 ret = pwrite(res->hr_localfd, hio->hio_data,
587 hio->hio_length,
588 hio->hio_offset + res->hr_localoff);
589 if (ret < 0)
590 hio->hio_error = errno;
591 else if (ret != (int64_t)hio->hio_length)
592 hio->hio_error = EIO;
593 else
594 hio->hio_error = 0;
595 break;
596 case HIO_DELETE:
597 ret = g_delete(res->hr_localfd,
598 hio->hio_offset + res->hr_localoff,
599 hio->hio_length);
600 if (ret < 0)
601 hio->hio_error = errno;
602 else
603 hio->hio_error = 0;
604 break;
605 case HIO_FLUSH:
606 ret = g_flush(res->hr_localfd);
607 if (ret < 0)
608 hio->hio_error = errno;
609 else
610 hio->hio_error = 0;
611 break;
612 }
613 if (hio->hio_error != 0) {
614 reqlog(LOG_ERR, 0, hio->hio_error, hio,
615 "Request failed: ");
616 }
617 pjdlog_debug(2, "disk: (%p) Moving request to the send queue.",
618 hio);
619 mtx_lock(&hio_send_list_lock);
620 wakeup = TAILQ_EMPTY(&hio_send_list);
621 TAILQ_INSERT_TAIL(&hio_send_list, hio, hio_next);
622 mtx_unlock(&hio_send_list_lock);
623 if (wakeup)
624 cv_signal(&hio_send_list_cond);
625 }
626 /* NOTREACHED */
627 return (NULL);
628}
629
630/*
631 * Thread sends requests back to primary node.
632 */
633static void *
634send_thread(void *arg)
635{
636 struct hast_resource *res = arg;
637 struct nv *nvout;
638 struct hio *hio;
639 void *data;
640 size_t length;
641 bool wakeup;
642
643 for (;;) {
644 pjdlog_debug(2, "send: Taking request.");
645 mtx_lock(&hio_send_list_lock);
646 while ((hio = TAILQ_FIRST(&hio_send_list)) == NULL) {
647 pjdlog_debug(2, "send: No requests, waiting.");
648 cv_wait(&hio_send_list_cond, &hio_send_list_lock);
649 }
650 TAILQ_REMOVE(&hio_send_list, hio, hio_next);
651 mtx_unlock(&hio_send_list_lock);
652 reqlog(LOG_DEBUG, 2, -1, hio, "send: (%p) Got request: ", hio);
653 nvout = nv_alloc();
654 /* Copy sequence number. */
655 nv_add_uint64(nvout, nv_get_uint64(hio->hio_nv, "seq"), "seq");
656 switch (hio->hio_cmd) {
657 case HIO_READ:
658 if (hio->hio_error == 0) {
659 data = hio->hio_data;
660 length = hio->hio_length;
661 break;
662 }
663 /*
664 * We send no data in case of an error.
665 */
666 /* FALLTHROUGH */
667 case HIO_DELETE:
668 case HIO_FLUSH:
669 case HIO_WRITE:
670 data = NULL;
671 length = 0;
672 break;
673 default:
674 abort();
675 break;
676 }
677 if (hio->hio_error != 0)
678 nv_add_int16(nvout, hio->hio_error, "error");
679 if (hast_proto_send(res, res->hr_remoteout, nvout, data,
680 length) < 0) {
681 pjdlog_exit(EX_TEMPFAIL, "Unable to send reply.");
682 }
683 nv_free(nvout);
684 pjdlog_debug(2, "disk: (%p) Moving request to the free queue.",
685 hio);
686 nv_free(hio->hio_nv);
687 hio->hio_error = 0;
688 mtx_lock(&hio_free_list_lock);
689 wakeup = TAILQ_EMPTY(&hio_free_list);
690 TAILQ_INSERT_TAIL(&hio_free_list, hio, hio_next);
691 mtx_unlock(&hio_free_list_lock);
692 if (wakeup)
693 cv_signal(&hio_free_list_cond);
694 }
695 /* NOTREACHED */
696 return (NULL);
697}
| 346 init_local(res);
347 init_remote(res, nvin);
348 init_environment();
349
350 error = pthread_create(&td, NULL, recv_thread, res);
351 assert(error == 0);
352 error = pthread_create(&td, NULL, disk_thread, res);
353 assert(error == 0);
354 error = pthread_create(&td, NULL, send_thread, res);
355 assert(error == 0);
356 (void)ctrl_thread(res);
357}
358
359static void
360reqlog(int loglevel, int debuglevel, int error, struct hio *hio, const char *fmt, ...)
361{
362 char msg[1024];
363 va_list ap;
364 int len;
365
366 va_start(ap, fmt);
367 len = vsnprintf(msg, sizeof(msg), fmt, ap);
368 va_end(ap);
369 if ((size_t)len < sizeof(msg)) {
370 switch (hio->hio_cmd) {
371 case HIO_READ:
372 (void)snprintf(msg + len, sizeof(msg) - len,
373 "READ(%ju, %ju).", (uintmax_t)hio->hio_offset,
374 (uintmax_t)hio->hio_length);
375 break;
376 case HIO_DELETE:
377 (void)snprintf(msg + len, sizeof(msg) - len,
378 "DELETE(%ju, %ju).", (uintmax_t)hio->hio_offset,
379 (uintmax_t)hio->hio_length);
380 break;
381 case HIO_FLUSH:
382 (void)snprintf(msg + len, sizeof(msg) - len, "FLUSH.");
383 break;
384 case HIO_WRITE:
385 (void)snprintf(msg + len, sizeof(msg) - len,
386 "WRITE(%ju, %ju).", (uintmax_t)hio->hio_offset,
387 (uintmax_t)hio->hio_length);
388 break;
389 default:
390 (void)snprintf(msg + len, sizeof(msg) - len,
391 "UNKNOWN(%u).", (unsigned int)hio->hio_cmd);
392 break;
393 }
394 }
395 pjdlog_common(loglevel, debuglevel, error, "%s", msg);
396}
397
398static int
399requnpack(struct hast_resource *res, struct hio *hio)
400{
401
402 hio->hio_cmd = nv_get_uint8(hio->hio_nv, "cmd");
403 if (hio->hio_cmd == 0) {
404 pjdlog_error("Header contains no 'cmd' field.");
405 hio->hio_error = EINVAL;
406 goto end;
407 }
408 switch (hio->hio_cmd) {
409 case HIO_READ:
410 case HIO_WRITE:
411 case HIO_DELETE:
412 hio->hio_offset = nv_get_uint64(hio->hio_nv, "offset");
413 if (nv_error(hio->hio_nv) != 0) {
414 pjdlog_error("Header is missing 'offset' field.");
415 hio->hio_error = EINVAL;
416 goto end;
417 }
418 hio->hio_length = nv_get_uint64(hio->hio_nv, "length");
419 if (nv_error(hio->hio_nv) != 0) {
420 pjdlog_error("Header is missing 'length' field.");
421 hio->hio_error = EINVAL;
422 goto end;
423 }
424 if (hio->hio_length == 0) {
425 pjdlog_error("Data length is zero.");
426 hio->hio_error = EINVAL;
427 goto end;
428 }
429 if (hio->hio_length > MAXPHYS) {
430 pjdlog_error("Data length is too large (%ju > %ju).",
431 (uintmax_t)hio->hio_length, (uintmax_t)MAXPHYS);
432 hio->hio_error = EINVAL;
433 goto end;
434 }
435 if ((hio->hio_offset % res->hr_local_sectorsize) != 0) {
436 pjdlog_error("Offset %ju is not multiple of sector size.",
437 (uintmax_t)hio->hio_offset);
438 hio->hio_error = EINVAL;
439 goto end;
440 }
441 if ((hio->hio_length % res->hr_local_sectorsize) != 0) {
442 pjdlog_error("Length %ju is not multiple of sector size.",
443 (uintmax_t)hio->hio_length);
444 hio->hio_error = EINVAL;
445 goto end;
446 }
447 if (hio->hio_offset + hio->hio_length >
448 (uint64_t)res->hr_datasize) {
449 pjdlog_error("Data offset is too large (%ju > %ju).",
450 (uintmax_t)(hio->hio_offset + hio->hio_length),
451 (uintmax_t)res->hr_datasize);
452 hio->hio_error = EINVAL;
453 goto end;
454 }
455 break;
456 default:
457 pjdlog_error("Header contains invalid 'cmd' (%hhu).",
458 hio->hio_cmd);
459 hio->hio_error = EINVAL;
460 goto end;
461 }
462 hio->hio_error = 0;
463end:
464 return (hio->hio_error);
465}
466
467/*
468 * Thread receives requests from the primary node.
469 */
470static void *
471recv_thread(void *arg)
472{
473 struct hast_resource *res = arg;
474 struct hio *hio;
475 bool wakeup;
476
477 for (;;) {
478 pjdlog_debug(2, "recv: Taking free request.");
479 mtx_lock(&hio_free_list_lock);
480 while ((hio = TAILQ_FIRST(&hio_free_list)) == NULL) {
481 pjdlog_debug(2, "recv: No free requests, waiting.");
482 cv_wait(&hio_free_list_cond, &hio_free_list_lock);
483 }
484 TAILQ_REMOVE(&hio_free_list, hio, hio_next);
485 mtx_unlock(&hio_free_list_lock);
486 pjdlog_debug(2, "recv: (%p) Got request.", hio);
487 if (hast_proto_recv_hdr(res->hr_remotein, &hio->hio_nv) < 0) {
488 pjdlog_exit(EX_TEMPFAIL,
489 "Unable to receive request header");
490 }
491 if (requnpack(res, hio) != 0)
492 goto send_queue;
493 reqlog(LOG_DEBUG, 2, -1, hio,
494 "recv: (%p) Got request header: ", hio);
495 if (hio->hio_cmd == HIO_WRITE) {
496 if (hast_proto_recv_data(res, res->hr_remotein,
497 hio->hio_nv, hio->hio_data, MAXPHYS) < 0) {
498 pjdlog_exit(EX_TEMPFAIL,
499 "Unable to receive reply data");
500 }
501 }
502 pjdlog_debug(2, "recv: (%p) Moving request to the disk queue.",
503 hio);
504 mtx_lock(&hio_disk_list_lock);
505 wakeup = TAILQ_EMPTY(&hio_disk_list);
506 TAILQ_INSERT_TAIL(&hio_disk_list, hio, hio_next);
507 mtx_unlock(&hio_disk_list_lock);
508 if (wakeup)
509 cv_signal(&hio_disk_list_cond);
510 continue;
511send_queue:
512 pjdlog_debug(2, "recv: (%p) Moving request to the send queue.",
513 hio);
514 mtx_lock(&hio_send_list_lock);
515 wakeup = TAILQ_EMPTY(&hio_send_list);
516 TAILQ_INSERT_TAIL(&hio_send_list, hio, hio_next);
517 mtx_unlock(&hio_send_list_lock);
518 if (wakeup)
519 cv_signal(&hio_send_list_cond);
520 }
521 /* NOTREACHED */
522 return (NULL);
523}
524
525/*
526 * Thread reads from or writes to local component and also handles DELETE and
527 * FLUSH requests.
528 */
529static void *
530disk_thread(void *arg)
531{
532 struct hast_resource *res = arg;
533 struct hio *hio;
534 ssize_t ret;
535 bool clear_activemap, wakeup;
536
537 clear_activemap = true;
538
539 for (;;) {
540 pjdlog_debug(2, "disk: Taking request.");
541 mtx_lock(&hio_disk_list_lock);
542 while ((hio = TAILQ_FIRST(&hio_disk_list)) == NULL) {
543 pjdlog_debug(2, "disk: No requests, waiting.");
544 cv_wait(&hio_disk_list_cond, &hio_disk_list_lock);
545 }
546 TAILQ_REMOVE(&hio_disk_list, hio, hio_next);
547 mtx_unlock(&hio_disk_list_lock);
548 while (clear_activemap) {
549 unsigned char *map;
550 size_t mapsize;
551
552 /*
553 * When first request is received, it means that primary
554 * already received our activemap, merged it and stored
555 * locally. We can now safely clear our activemap.
556 */
557 mapsize =
558 activemap_calc_ondisk_size(res->hr_local_mediasize -
559 METADATA_SIZE, res->hr_extentsize,
560 res->hr_local_sectorsize);
561 map = calloc(1, mapsize);
562 if (map == NULL) {
563 pjdlog_warning("Unable to allocate memory to clear local activemap.");
564 break;
565 }
566 if (pwrite(res->hr_localfd, map, mapsize,
567 METADATA_SIZE) != (ssize_t)mapsize) {
568 pjdlog_errno(LOG_WARNING,
569 "Unable to store cleared activemap");
570 free(map);
571 break;
572 }
573 free(map);
574 clear_activemap = false;
575 pjdlog_debug(1, "Local activemap cleared.");
576 }
577 reqlog(LOG_DEBUG, 2, -1, hio, "disk: (%p) Got request: ", hio);
578 /* Handle the actual request. */
579 switch (hio->hio_cmd) {
580 case HIO_READ:
581 ret = pread(res->hr_localfd, hio->hio_data,
582 hio->hio_length,
583 hio->hio_offset + res->hr_localoff);
584 if (ret < 0)
585 hio->hio_error = errno;
586 else if (ret != (int64_t)hio->hio_length)
587 hio->hio_error = EIO;
588 else
589 hio->hio_error = 0;
590 break;
591 case HIO_WRITE:
592 ret = pwrite(res->hr_localfd, hio->hio_data,
593 hio->hio_length,
594 hio->hio_offset + res->hr_localoff);
595 if (ret < 0)
596 hio->hio_error = errno;
597 else if (ret != (int64_t)hio->hio_length)
598 hio->hio_error = EIO;
599 else
600 hio->hio_error = 0;
601 break;
602 case HIO_DELETE:
603 ret = g_delete(res->hr_localfd,
604 hio->hio_offset + res->hr_localoff,
605 hio->hio_length);
606 if (ret < 0)
607 hio->hio_error = errno;
608 else
609 hio->hio_error = 0;
610 break;
611 case HIO_FLUSH:
612 ret = g_flush(res->hr_localfd);
613 if (ret < 0)
614 hio->hio_error = errno;
615 else
616 hio->hio_error = 0;
617 break;
618 }
619 if (hio->hio_error != 0) {
620 reqlog(LOG_ERR, 0, hio->hio_error, hio,
621 "Request failed: ");
622 }
623 pjdlog_debug(2, "disk: (%p) Moving request to the send queue.",
624 hio);
625 mtx_lock(&hio_send_list_lock);
626 wakeup = TAILQ_EMPTY(&hio_send_list);
627 TAILQ_INSERT_TAIL(&hio_send_list, hio, hio_next);
628 mtx_unlock(&hio_send_list_lock);
629 if (wakeup)
630 cv_signal(&hio_send_list_cond);
631 }
632 /* NOTREACHED */
633 return (NULL);
634}
635
636/*
637 * Thread sends requests back to primary node.
638 */
639static void *
640send_thread(void *arg)
641{
642 struct hast_resource *res = arg;
643 struct nv *nvout;
644 struct hio *hio;
645 void *data;
646 size_t length;
647 bool wakeup;
648
649 for (;;) {
650 pjdlog_debug(2, "send: Taking request.");
651 mtx_lock(&hio_send_list_lock);
652 while ((hio = TAILQ_FIRST(&hio_send_list)) == NULL) {
653 pjdlog_debug(2, "send: No requests, waiting.");
654 cv_wait(&hio_send_list_cond, &hio_send_list_lock);
655 }
656 TAILQ_REMOVE(&hio_send_list, hio, hio_next);
657 mtx_unlock(&hio_send_list_lock);
658 reqlog(LOG_DEBUG, 2, -1, hio, "send: (%p) Got request: ", hio);
659 nvout = nv_alloc();
660 /* Copy sequence number. */
661 nv_add_uint64(nvout, nv_get_uint64(hio->hio_nv, "seq"), "seq");
662 switch (hio->hio_cmd) {
663 case HIO_READ:
664 if (hio->hio_error == 0) {
665 data = hio->hio_data;
666 length = hio->hio_length;
667 break;
668 }
669 /*
670 * We send no data in case of an error.
671 */
672 /* FALLTHROUGH */
673 case HIO_DELETE:
674 case HIO_FLUSH:
675 case HIO_WRITE:
676 data = NULL;
677 length = 0;
678 break;
679 default:
680 abort();
681 break;
682 }
683 if (hio->hio_error != 0)
684 nv_add_int16(nvout, hio->hio_error, "error");
685 if (hast_proto_send(res, res->hr_remoteout, nvout, data,
686 length) < 0) {
687 pjdlog_exit(EX_TEMPFAIL, "Unable to send reply.");
688 }
689 nv_free(nvout);
690 pjdlog_debug(2, "disk: (%p) Moving request to the free queue.",
691 hio);
692 nv_free(hio->hio_nv);
693 hio->hio_error = 0;
694 mtx_lock(&hio_free_list_lock);
695 wakeup = TAILQ_EMPTY(&hio_free_list);
696 TAILQ_INSERT_TAIL(&hio_free_list, hio, hio_next);
697 mtx_unlock(&hio_free_list_lock);
698 if (wakeup)
699 cv_signal(&hio_free_list_cond);
700 }
701 /* NOTREACHED */
702 return (NULL);
703}
|