hammer2 - Performance work
[dragonfly.git] / sys / vfs / hammer2 / hammer2_admin.c
1 /*
2  * Copyright (c) 2015-2018 The DragonFly Project.  All rights reserved.
3  *
4  * This code is derived from software contributed to The DragonFly Project
5  * by Matthew Dillon <dillon@dragonflybsd.org>
6  *
7  * Redistribution and use in source and binary forms, with or without
8  * modification, are permitted provided that the following conditions
9  * are met:
10  *
11  * 1. Redistributions of source code must retain the above copyright
12  *    notice, this list of conditions and the following disclaimer.
13  * 2. Redistributions in binary form must reproduce the above copyright
14  *    notice, this list of conditions and the following disclaimer in
15  *    the documentation and/or other materials provided with the
16  *    distribution.
17  * 3. Neither the name of The DragonFly Project nor the names of its
18  *    contributors may be used to endorse or promote products derived
19  *    from this software without specific, prior written permission.
20  *
21  * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
22  * ``AS IS'' AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
23  * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS
24  * FOR A PARTICULAR PURPOSE ARE DISCLAIMED.  IN NO EVENT SHALL THE
25  * COPYRIGHT HOLDERS OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT,
26  * INCIDENTAL, SPECIAL, EXEMPLARY OR CONSEQUENTIAL DAMAGES (INCLUDING,
27  * BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
28  * LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED
29  * AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
30  * OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT
31  * OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
32  * SUCH DAMAGE.
33  */
34 /*
35  * This module implements the hammer2 helper thread API, including
36  * the frontend/backend XOP API.
37  */
38 #include "hammer2.h"
39
40 #define H2XOPDESCRIPTOR(label)                                  \
41         hammer2_xop_desc_t hammer2_##label##_desc = {           \
42                 .storage_func = hammer2_xop_##label,            \
43                 .id = #label                                    \
44         };                                                      \
45
46 H2XOPDESCRIPTOR(ipcluster);
47 H2XOPDESCRIPTOR(readdir);
48 H2XOPDESCRIPTOR(nresolve);
49 H2XOPDESCRIPTOR(unlink);
50 H2XOPDESCRIPTOR(nrename);
51 H2XOPDESCRIPTOR(scanlhc);
52 H2XOPDESCRIPTOR(scanall);
53 H2XOPDESCRIPTOR(lookup);
54 H2XOPDESCRIPTOR(delete);
55 H2XOPDESCRIPTOR(inode_mkdirent);
56 H2XOPDESCRIPTOR(inode_create);
57 H2XOPDESCRIPTOR(inode_create_det);
58 H2XOPDESCRIPTOR(inode_create_ins);
59 H2XOPDESCRIPTOR(inode_destroy);
60 H2XOPDESCRIPTOR(inode_chain_sync);
61 H2XOPDESCRIPTOR(inode_unlinkall);
62 H2XOPDESCRIPTOR(inode_connect);
63 H2XOPDESCRIPTOR(inode_flush);
64 H2XOPDESCRIPTOR(strategy_read);
65 H2XOPDESCRIPTOR(strategy_write);
66
67 /*
68  * Set flags and wakeup any waiters.
69  *
70  * WARNING! During teardown (thr) can disappear the instant our cmpset
71  *          succeeds.
72  */
73 void
74 hammer2_thr_signal(hammer2_thread_t *thr, uint32_t flags)
75 {
76         uint32_t oflags;
77         uint32_t nflags;
78
79         for (;;) {
80                 oflags = thr->flags;
81                 cpu_ccfence();
82                 nflags = (oflags | flags) & ~HAMMER2_THREAD_WAITING;
83
84                 if (oflags & HAMMER2_THREAD_WAITING) {
85                         if (atomic_cmpset_int(&thr->flags, oflags, nflags)) {
86                                 wakeup(&thr->flags);
87                                 break;
88                         }
89                 } else {
90                         if (atomic_cmpset_int(&thr->flags, oflags, nflags))
91                                 break;
92                 }
93         }
94 }
95
96 /*
97  * Set and clear flags and wakeup any waiters.
98  *
99  * WARNING! During teardown (thr) can disappear the instant our cmpset
100  *          succeeds.
101  */
102 void
103 hammer2_thr_signal2(hammer2_thread_t *thr, uint32_t posflags, uint32_t negflags)
104 {
105         uint32_t oflags;
106         uint32_t nflags;
107
108         for (;;) {
109                 oflags = thr->flags;
110                 cpu_ccfence();
111                 nflags = (oflags | posflags) &
112                         ~(negflags | HAMMER2_THREAD_WAITING);
113                 if (oflags & HAMMER2_THREAD_WAITING) {
114                         if (atomic_cmpset_int(&thr->flags, oflags, nflags)) {
115                                 wakeup(&thr->flags);
116                                 break;
117                         }
118                 } else {
119                         if (atomic_cmpset_int(&thr->flags, oflags, nflags))
120                                 break;
121                 }
122         }
123 }
124
125 /*
126  * Wait until all the bits in flags are set.
127  *
128  * WARNING! During teardown (thr) can disappear the instant our cmpset
129  *          succeeds.
130  */
131 void
132 hammer2_thr_wait(hammer2_thread_t *thr, uint32_t flags)
133 {
134         uint32_t oflags;
135         uint32_t nflags;
136
137         for (;;) {
138                 oflags = thr->flags;
139                 cpu_ccfence();
140                 if ((oflags & flags) == flags)
141                         break;
142                 nflags = oflags | HAMMER2_THREAD_WAITING;
143                 tsleep_interlock(&thr->flags, 0);
144                 if (atomic_cmpset_int(&thr->flags, oflags, nflags)) {
145                         tsleep(&thr->flags, PINTERLOCKED, "h2twait", hz*60);
146                 }
147         }
148 }
149
150 /*
151  * Wait until any of the bits in flags are set, with timeout.
152  *
153  * WARNING! During teardown (thr) can disappear the instant our cmpset
154  *          succeeds.
155  */
156 int
157 hammer2_thr_wait_any(hammer2_thread_t *thr, uint32_t flags, int timo)
158 {
159         uint32_t oflags;
160         uint32_t nflags;
161         int error;
162
163         error = 0;
164         for (;;) {
165                 oflags = thr->flags;
166                 cpu_ccfence();
167                 if (oflags & flags)
168                         break;
169                 nflags = oflags | HAMMER2_THREAD_WAITING;
170                 tsleep_interlock(&thr->flags, 0);
171                 if (atomic_cmpset_int(&thr->flags, oflags, nflags)) {
172                         error = tsleep(&thr->flags, PINTERLOCKED,
173                                        "h2twait", timo);
174                 }
175                 if (error == ETIMEDOUT) {
176                         error = HAMMER2_ERROR_ETIMEDOUT;
177                         break;
178                 }
179         }
180         return error;
181 }
182
183 /*
184  * Wait until the bits in flags are clear.
185  *
186  * WARNING! During teardown (thr) can disappear the instant our cmpset
187  *          succeeds.
188  */
189 void
190 hammer2_thr_wait_neg(hammer2_thread_t *thr, uint32_t flags)
191 {
192         uint32_t oflags;
193         uint32_t nflags;
194
195         for (;;) {
196                 oflags = thr->flags;
197                 cpu_ccfence();
198                 if ((oflags & flags) == 0)
199                         break;
200                 nflags = oflags | HAMMER2_THREAD_WAITING;
201                 tsleep_interlock(&thr->flags, 0);
202                 if (atomic_cmpset_int(&thr->flags, oflags, nflags)) {
203                         tsleep(&thr->flags, PINTERLOCKED, "h2twait", hz*60);
204                 }
205         }
206 }
207
208 /*
209  * Initialize the supplied thread structure, starting the specified
210  * thread.
211  *
212  * NOTE: thr structure can be retained across mounts and unmounts for this
213  *       pmp, so make sure the flags are in a sane state.
214  */
215 void
216 hammer2_thr_create(hammer2_thread_t *thr, hammer2_pfs_t *pmp,
217                    hammer2_dev_t *hmp,
218                    const char *id, int clindex, int repidx,
219                    void (*func)(void *arg))
220 {
221         thr->pmp = pmp;         /* xop helpers */
222         thr->hmp = hmp;         /* bulkfree */
223         thr->clindex = clindex;
224         thr->repidx = repidx;
225         TAILQ_INIT(&thr->xopq);
226         atomic_clear_int(&thr->flags, HAMMER2_THREAD_STOP |
227                                       HAMMER2_THREAD_STOPPED |
228                                       HAMMER2_THREAD_FREEZE |
229                                       HAMMER2_THREAD_FROZEN);
230         if (thr->scratch == NULL)
231                 thr->scratch = kmalloc(MAXPHYS, M_HAMMER2, M_WAITOK | M_ZERO);
232         if (repidx >= 0) {
233                 lwkt_create(func, thr, &thr->td, NULL, 0, repidx % ncpus,
234                             "%s-%s.%02d", id, pmp->pfs_names[clindex], repidx);
235         } else if (pmp) {
236                 lwkt_create(func, thr, &thr->td, NULL, 0, -1,
237                             "%s-%s", id, pmp->pfs_names[clindex]);
238         } else {
239                 lwkt_create(func, thr, &thr->td, NULL, 0, -1, "%s", id);
240         }
241 }
242
243 /*
244  * Terminate a thread.  This function will silently return if the thread
245  * was never initialized or has already been deleted.
246  *
247  * This is accomplished by setting the STOP flag and waiting for the td
248  * structure to become NULL.
249  */
250 void
251 hammer2_thr_delete(hammer2_thread_t *thr)
252 {
253         if (thr->td == NULL)
254                 return;
255         hammer2_thr_signal(thr, HAMMER2_THREAD_STOP);
256         hammer2_thr_wait(thr, HAMMER2_THREAD_STOPPED);
257         thr->pmp = NULL;
258         if (thr->scratch) {
259                 kfree(thr->scratch, M_HAMMER2);
260                 thr->scratch = NULL;
261         }
262         KKASSERT(TAILQ_EMPTY(&thr->xopq));
263 }
264
265 /*
266  * Asynchronous remaster request.  Ask the synchronization thread to
267  * start over soon (as if it were frozen and unfrozen, but without waiting).
268  * The thread always recalculates mastership relationships when restarting.
269  */
270 void
271 hammer2_thr_remaster(hammer2_thread_t *thr)
272 {
273         if (thr->td == NULL)
274                 return;
275         hammer2_thr_signal(thr, HAMMER2_THREAD_REMASTER);
276 }
277
278 void
279 hammer2_thr_freeze_async(hammer2_thread_t *thr)
280 {
281         hammer2_thr_signal(thr, HAMMER2_THREAD_FREEZE);
282 }
283
284 void
285 hammer2_thr_freeze(hammer2_thread_t *thr)
286 {
287         if (thr->td == NULL)
288                 return;
289         hammer2_thr_signal(thr, HAMMER2_THREAD_FREEZE);
290         hammer2_thr_wait(thr, HAMMER2_THREAD_FROZEN);
291 }
292
293 void
294 hammer2_thr_unfreeze(hammer2_thread_t *thr)
295 {
296         if (thr->td == NULL)
297                 return;
298         hammer2_thr_signal(thr, HAMMER2_THREAD_UNFREEZE);
299         hammer2_thr_wait_neg(thr, HAMMER2_THREAD_FROZEN);
300 }
301
302 int
303 hammer2_thr_break(hammer2_thread_t *thr)
304 {
305         if (thr->flags & (HAMMER2_THREAD_STOP |
306                           HAMMER2_THREAD_REMASTER |
307                           HAMMER2_THREAD_FREEZE)) {
308                 return 1;
309         }
310         return 0;
311 }
312
313 /****************************************************************************
314  *                          HAMMER2 XOPS API                                *
315  ****************************************************************************/
316
317 void
318 hammer2_xop_group_init(hammer2_pfs_t *pmp, hammer2_xop_group_t *xgrp)
319 {
320         /* no extra fields in structure at the moment */
321 }
322
323 /*
324  * Allocate a XOP request.
325  *
326  * Once allocated a XOP request can be started, collected, and retired,
327  * and can be retired early if desired.
328  *
329  * NOTE: Fifo indices might not be zero but ri == wi on objcache_get().
330  */
331 void *
332 hammer2_xop_alloc(hammer2_inode_t *ip, int flags)
333 {
334         hammer2_xop_t *xop;
335
336         xop = objcache_get(cache_xops, M_WAITOK);
337         KKASSERT(xop->head.cluster.array[0].chain == NULL);
338
339         xop->head.ip1 = ip;
340         xop->head.desc = NULL;
341         xop->head.flags = flags;
342         xop->head.state = 0;
343         xop->head.error = 0;
344         xop->head.collect_key = 0;
345         xop->head.focus_dio = NULL;
346
347         if (flags & HAMMER2_XOP_MODIFYING)
348                 xop->head.mtid = hammer2_trans_sub(ip->pmp);
349         else
350                 xop->head.mtid = 0;
351
352         xop->head.cluster.nchains = ip->cluster.nchains;
353         xop->head.cluster.pmp = ip->pmp;
354         xop->head.cluster.flags = HAMMER2_CLUSTER_LOCKED;
355
356         /*
357          * run_mask - Active thread (or frontend) associated with XOP
358          */
359         xop->head.run_mask = HAMMER2_XOPMASK_VOP;
360
361         hammer2_inode_ref(ip);
362
363         return xop;
364 }
365
366 void
367 hammer2_xop_setname(hammer2_xop_head_t *xop, const char *name, size_t name_len)
368 {
369         xop->name1 = kmalloc(name_len + 1, M_HAMMER2, M_WAITOK | M_ZERO);
370         xop->name1_len = name_len;
371         bcopy(name, xop->name1, name_len);
372 }
373
374 void
375 hammer2_xop_setname2(hammer2_xop_head_t *xop, const char *name, size_t name_len)
376 {
377         xop->name2 = kmalloc(name_len + 1, M_HAMMER2, M_WAITOK | M_ZERO);
378         xop->name2_len = name_len;
379         bcopy(name, xop->name2, name_len);
380 }
381
382 size_t
383 hammer2_xop_setname_inum(hammer2_xop_head_t *xop, hammer2_key_t inum)
384 {
385         const size_t name_len = 18;
386
387         xop->name1 = kmalloc(name_len + 1, M_HAMMER2, M_WAITOK | M_ZERO);
388         xop->name1_len = name_len;
389         ksnprintf(xop->name1, name_len + 1, "0x%016jx", (intmax_t)inum);
390
391         return name_len;
392 }
393
394
395 void
396 hammer2_xop_setip2(hammer2_xop_head_t *xop, hammer2_inode_t *ip2)
397 {
398         xop->ip2 = ip2;
399         hammer2_inode_ref(ip2);
400 }
401
402 void
403 hammer2_xop_setip3(hammer2_xop_head_t *xop, hammer2_inode_t *ip3)
404 {
405         xop->ip3 = ip3;
406         hammer2_inode_ref(ip3);
407 }
408
409 void
410 hammer2_xop_reinit(hammer2_xop_head_t *xop)
411 {
412         xop->state = 0;
413         xop->error = 0;
414         xop->collect_key = 0;
415         xop->run_mask = HAMMER2_XOPMASK_VOP;
416 }
417
418 /*
419  * A mounted PFS needs Xops threads to support frontend operations.
420  */
421 void
422 hammer2_xop_helper_create(hammer2_pfs_t *pmp)
423 {
424         int i;
425         int j;
426
427         lockmgr(&pmp->lock, LK_EXCLUSIVE);
428         pmp->has_xop_threads = 1;
429
430         for (i = 0; i < pmp->iroot->cluster.nchains; ++i) {
431                 for (j = 0; j < HAMMER2_XOPGROUPS; ++j) {
432                         if (pmp->xop_groups[j].thrs[i].td)
433                                 continue;
434                         hammer2_thr_create(&pmp->xop_groups[j].thrs[i],
435                                            pmp, NULL,
436                                            "h2xop", i, j,
437                                            hammer2_primary_xops_thread);
438                 }
439         }
440         lockmgr(&pmp->lock, LK_RELEASE);
441 }
442
443 void
444 hammer2_xop_helper_cleanup(hammer2_pfs_t *pmp)
445 {
446         int i;
447         int j;
448
449         for (i = 0; i < pmp->pfs_nmasters; ++i) {
450                 for (j = 0; j < HAMMER2_XOPGROUPS; ++j) {
451                         if (pmp->xop_groups[j].thrs[i].td)
452                                 hammer2_thr_delete(&pmp->xop_groups[j].thrs[i]);
453                 }
454         }
455         pmp->has_xop_threads = 0;
456 }
457
458 /*
459  * Start a XOP request, queueing it to all nodes in the cluster to
460  * execute the cluster op.
461  *
462  * XXX optimize single-target case.
463  */
464 void
465 hammer2_xop_start_except(hammer2_xop_head_t *xop, hammer2_xop_desc_t *desc,
466                          int notidx)
467 {
468         hammer2_inode_t *ip1;
469         hammer2_pfs_t *pmp;
470         hammer2_thread_t *thr;
471         int i;
472         int ng;
473         int nchains;
474
475         ip1 = xop->ip1;
476         pmp = ip1->pmp;
477         if (pmp->has_xop_threads == 0)
478                 hammer2_xop_helper_create(pmp);
479
480         /*
481          * The intent of the XOP sequencer is to ensure that ops on the same
482          * inode execute in the same order.  This is necessary when issuing
483          * modifying operations to multiple targets because some targets might
484          * get behind and the frontend is allowed to complete the moment a
485          * quorum of targets succeed.
486          *
487          * Strategy operations:
488          *
489          *      (1) Must be segregated from non-strategy operations to
490          *          avoid a deadlock.  A vfsync and a bread/bwrite can
491          *          deadlock the vfsync's buffer list scan.
492          *
493          *      (2) Reads are separated from writes to avoid write stalls
494          *          from excessively intefering with reads.  Reads are allowed
495          *          to wander across multiple worker threads for potential
496          *          single-file concurrency improvements.
497          *
498          *      (3) Writes are serialized to a single worker thread (for any
499          *          given inode) in order to try to improve block allocation
500          *          sequentiality and to reduce lock contention.
501          *
502          * TODO - RENAME fails here because it is potentially modifying
503          *        three different inodes, but we triple-lock the inodes
504          *        involved so it shouldn't create a sequencing schism.
505          */
506         if (xop->flags & HAMMER2_XOP_STRATEGY) {
507                 hammer2_xop_strategy_t *xopst;
508                 hammer2_off_t off;
509                 int cdr;
510
511                 xopst = &((hammer2_xop_t *)xop)->xop_strategy;
512                 ng = (int)(hammer2_icrc32(&xop->ip1, sizeof(xop->ip1)));
513                 if (desc == &hammer2_strategy_read_desc) {
514                         off = xopst->lbase / HAMMER2_PBUFSIZE;
515                         cdr = hammer2_cluster_data_read;
516                         /* sysctl race, load into var */
517                         cpu_ccfence();
518                         if (cdr)
519                                 off /= cdr;
520                         ng ^= hammer2_icrc32(&off, sizeof(off)) &
521                               (hammer2_worker_rmask << 1);
522                         ng |= 1;
523                 } else {
524 #if 0
525                         off = xopst->lbase >> 21;
526                         ng ^= hammer2_icrc32(&off, sizeof(off)) & 3;
527 #endif
528                         ng &= ~1;
529                 }
530                 ng = ng & (HAMMER2_XOPGROUPS_MASK >> 1);
531                 ng += HAMMER2_XOPGROUPS / 2;
532         } else {
533                 ng = (int)(hammer2_icrc32(&xop->ip1, sizeof(xop->ip1)));
534                 ng = ng & (HAMMER2_XOPGROUPS_MASK >> 1);
535         }
536         xop->desc = desc;
537
538         /*
539          * The instant xop is queued another thread can pick it off.  In the
540          * case of asynchronous ops, another thread might even finish and
541          * deallocate it.
542          */
543         hammer2_spin_ex(&pmp->xop_spin);
544         nchains = ip1->cluster.nchains;
545         for (i = 0; i < nchains; ++i) {
546                 /*
547                  * XXX ip1->cluster.array* not stable here.  This temporary
548                  *     hack fixes basic issues in target XOPs which need to
549                  *     obtain a starting chain from the inode but does not
550                  *     address possible races against inode updates which
551                  *     might NULL-out a chain.
552                  */
553                 if (i != notidx && ip1->cluster.array[i].chain) {
554                         thr = &pmp->xop_groups[ng].thrs[i];
555                         atomic_set_64(&xop->run_mask, 1LLU << i);
556                         atomic_set_64(&xop->chk_mask, 1LLU << i);
557                         xop->collect[i].thr = thr;
558                         TAILQ_INSERT_TAIL(&thr->xopq, xop, collect[i].entry);
559                 }
560         }
561         hammer2_spin_unex(&pmp->xop_spin);
562         /* xop can become invalid at this point */
563
564         /*
565          * Each thread has its own xopq
566          */
567         for (i = 0; i < nchains; ++i) {
568                 if (i != notidx) {
569                         thr = &pmp->xop_groups[ng].thrs[i];
570                         hammer2_thr_signal(thr, HAMMER2_THREAD_XOPQ);
571                 }
572         }
573 }
574
575 void
576 hammer2_xop_start(hammer2_xop_head_t *xop, hammer2_xop_desc_t *desc)
577 {
578         hammer2_xop_start_except(xop, desc, -1);
579 }
580
581 /*
582  * Retire a XOP.  Used by both the VOP frontend and by the XOP backend.
583  */
584 void
585 hammer2_xop_retire(hammer2_xop_head_t *xop, uint64_t mask)
586 {
587         hammer2_chain_t *chain;
588         uint64_t nmask;
589         int i;
590
591         /*
592          * Remove the frontend collector or remove a backend feeder.
593          *
594          * When removing the frontend we must wakeup any backend feeders
595          * who are waiting for FIFO space.
596          *
597          * When removing the last backend feeder we must wakeup any waiting
598          * frontend.
599          */
600         KKASSERT(xop->run_mask & mask);
601         nmask = atomic_fetchadd_64(&xop->run_mask,
602                                    -mask + HAMMER2_XOPMASK_FEED);
603
604         /*
605          * More than one entity left
606          */
607         if ((nmask & HAMMER2_XOPMASK_ALLDONE) != mask) {
608                 /*
609                  * Frontend terminating, wakeup any backends waiting on
610                  * fifo full.
611                  *
612                  * NOTE!!! The xop can get ripped out from under us at
613                  *         this point, so do not reference it again.
614                  *         The wakeup(xop) doesn't touch the xop and
615                  *         is ok.
616                  */
617                 if (mask == HAMMER2_XOPMASK_VOP) {
618                         if (nmask & HAMMER2_XOPMASK_FIFOW)
619                                 wakeup(xop);
620                 }
621
622                 /*
623                  * Wakeup frontend if the last backend is terminating.
624                  */
625                 nmask -= mask;
626                 if ((nmask & HAMMER2_XOPMASK_ALLDONE) == HAMMER2_XOPMASK_VOP) {
627                         if (nmask & HAMMER2_XOPMASK_WAIT)
628                                 wakeup(xop);
629                 }
630
631                 return;
632         }
633         /* else nobody else left, we can ignore FIFOW */
634
635         /*
636          * All collectors are gone, we can cleanup and dispose of the XOP.
637          * Note that this can wind up being a frontend OR a backend.
638          * Pending chains are locked shared and not owned by any thread.
639          *
640          * Cleanup the collection cluster.
641          */
642         for (i = 0; i < xop->cluster.nchains; ++i) {
643                 xop->cluster.array[i].flags = 0;
644                 chain = xop->cluster.array[i].chain;
645                 if (chain) {
646                         xop->cluster.array[i].chain = NULL;
647                         hammer2_chain_drop_unhold(chain);
648                 }
649         }
650
651         /*
652          * Cleanup the fifos.  Since we are the only entity left on this
653          * xop we don't have to worry about fifo flow control, and one
654          * lfence() will do the job.
655          */
656         cpu_lfence();
657         mask = xop->chk_mask;
658         for (i = 0; mask && i < HAMMER2_MAXCLUSTER; ++i) {
659                 hammer2_xop_fifo_t *fifo = &xop->collect[i];
660                 while (fifo->ri != fifo->wi) {
661                         chain = fifo->array[fifo->ri & HAMMER2_XOPFIFO_MASK];
662                         if (chain)
663                                 hammer2_chain_drop_unhold(chain);
664                         ++fifo->ri;
665                 }
666                 mask &= ~(1U << i);
667         }
668
669         /*
670          * The inode is only held at this point, simply drop it.
671          */
672         if (xop->ip1) {
673                 hammer2_inode_drop(xop->ip1);
674                 xop->ip1 = NULL;
675         }
676         if (xop->ip2) {
677                 hammer2_inode_drop(xop->ip2);
678                 xop->ip2 = NULL;
679         }
680         if (xop->ip3) {
681                 hammer2_inode_drop(xop->ip3);
682                 xop->ip3 = NULL;
683         }
684         if (xop->name1) {
685                 kfree(xop->name1, M_HAMMER2);
686                 xop->name1 = NULL;
687                 xop->name1_len = 0;
688         }
689         if (xop->name2) {
690                 kfree(xop->name2, M_HAMMER2);
691                 xop->name2 = NULL;
692                 xop->name2_len = 0;
693         }
694
695         objcache_put(cache_xops, xop);
696 }
697
698 /*
699  * (Backend) Returns non-zero if the frontend is still attached.
700  */
701 int
702 hammer2_xop_active(hammer2_xop_head_t *xop)
703 {
704         if (xop->run_mask & HAMMER2_XOPMASK_VOP)
705                 return 1;
706         else
707                 return 0;
708 }
709
710 /*
711  * (Backend) Feed chain data through the cluster validator and back to
712  * the frontend.  Chains are fed from multiple nodes concurrently
713  * and pipelined via per-node FIFOs in the XOP.
714  *
715  * The chain must be locked (either shared or exclusive).  The caller may
716  * unlock and drop the chain on return.  This function will add an extra
717  * ref and hold the chain's data for the pass-back.
718  *
719  * No xop lock is needed because we are only manipulating fields under
720  * our direct control.
721  *
722  * Returns 0 on success and a hammer2 error code if sync is permanently
723  * lost.  The caller retains a ref on the chain but by convention
724  * the lock is typically inherited by the xop (caller loses lock).
725  *
726  * Returns non-zero on error.  In this situation the caller retains a
727  * ref on the chain but loses the lock (we unlock here).
728  */
729 int
730 hammer2_xop_feed(hammer2_xop_head_t *xop, hammer2_chain_t *chain,
731                  int clindex, int error)
732 {
733         hammer2_xop_fifo_t *fifo;
734         uint64_t mask;
735
736         /*
737          * Early termination (typicaly of xop_readir)
738          */
739         if (hammer2_xop_active(xop) == 0) {
740                 error = HAMMER2_ERROR_ABORTED;
741                 goto done;
742         }
743
744         /*
745          * Multi-threaded entry into the XOP collector.  We own the
746          * fifo->wi for our clindex.
747          */
748         fifo = &xop->collect[clindex];
749
750         if (fifo->ri == fifo->wi - HAMMER2_XOPFIFO)
751                 lwkt_yield();
752         while (fifo->ri == fifo->wi - HAMMER2_XOPFIFO) {
753                 atomic_set_int(&fifo->flags, HAMMER2_XOP_FIFO_STALL);
754                 mask = xop->run_mask;
755                 if ((mask & HAMMER2_XOPMASK_VOP) == 0) {
756                         error = HAMMER2_ERROR_ABORTED;
757                         goto done;
758                 }
759                 tsleep_interlock(xop, 0);
760                 if (atomic_cmpset_64(&xop->run_mask, mask,
761                                      mask | HAMMER2_XOPMASK_FIFOW)) {
762                         if (fifo->ri == fifo->wi - HAMMER2_XOPFIFO) {
763                                 tsleep(xop, PINTERLOCKED, "h2feed", hz*60);
764                         }
765                 }
766                 /* retry */
767         }
768         atomic_clear_int(&fifo->flags, HAMMER2_XOP_FIFO_STALL);
769         if (chain)
770                 hammer2_chain_ref_hold(chain);
771         if (error == 0 && chain)
772                 error = chain->error;
773         fifo->errors[fifo->wi & HAMMER2_XOPFIFO_MASK] = error;
774         fifo->array[fifo->wi & HAMMER2_XOPFIFO_MASK] = chain;
775         cpu_sfence();
776         ++fifo->wi;
777
778         mask = atomic_fetchadd_64(&xop->run_mask, HAMMER2_XOPMASK_FEED);
779         if (mask & HAMMER2_XOPMASK_WAIT) {
780                 atomic_clear_64(&xop->run_mask, HAMMER2_XOPMASK_WAIT);
781                 wakeup(xop);
782         }
783         error = 0;
784
785         /*
786          * Cleanup.  If an error occurred we eat the lock.  If no error
787          * occurred the fifo inherits the lock and gains an additional ref.
788          *
789          * The caller's ref remains in both cases.
790          */
791 done:
792         return error;
793 }
794
795 /*
796  * (Frontend) collect a response from a running cluster op.
797  *
798  * Responses are fed from all appropriate nodes concurrently
799  * and collected into a cohesive response >= collect_key.
800  *
801  * The collector will return the instant quorum or other requirements
802  * are met, even if some nodes get behind or become non-responsive.
803  *
804  * HAMMER2_XOP_COLLECT_NOWAIT   - Used to 'poll' a completed collection,
805  *                                usually called synchronously from the
806  *                                node XOPs for the strategy code to
807  *                                fake the frontend collection and complete
808  *                                the BIO as soon as possible.
809  *
810  * HAMMER2_XOP_SYNCHRONIZER     - Reqeuest synchronization with a particular
811  *                                cluster index, prevents looping when that
812  *                                index is out of sync so caller can act on
813  *                                the out of sync element.  ESRCH and EDEADLK
814  *                                can be returned if this flag is specified.
815  *
816  * Returns 0 on success plus a filled out xop->cluster structure.
817  * Return ENOENT on normal termination.
818  * Otherwise return an error.
819  *
820  * WARNING! If the xop returns a cluster with a non-NULL focus, note that
821  *          none of the chains in the cluster (or the focus) are either
822  *          locked or I/O synchronized with the cpu.  hammer2_xop_gdata()
823  *          and hammer2_xop_pdata() must be used to safely access the focus
824  *          chain's content.
825  *
826  *          The frontend can make certain assumptions based on higher-level
827  *          locking done by the frontend, but data integrity absolutely
828  *          requires using the gdata/pdata API.
829  */
830 int
831 hammer2_xop_collect(hammer2_xop_head_t *xop, int flags)
832 {
833         hammer2_xop_fifo_t *fifo;
834         hammer2_chain_t *chain;
835         hammer2_key_t lokey;
836         uint64_t mask;
837         int error;
838         int keynull;
839         int adv;                /* advance the element */
840         int i;
841
842 loop:
843         /*
844          * First loop tries to advance pieces of the cluster which
845          * are out of sync.
846          */
847         lokey = HAMMER2_KEY_MAX;
848         keynull = HAMMER2_CHECK_NULL;
849         mask = xop->run_mask;
850         cpu_lfence();
851
852         for (i = 0; i < xop->cluster.nchains; ++i) {
853                 chain = xop->cluster.array[i].chain;
854                 if (chain == NULL) {
855                         adv = 1;
856                 } else if (chain->bref.key < xop->collect_key) {
857                         adv = 1;
858                 } else {
859                         keynull &= ~HAMMER2_CHECK_NULL;
860                         if (lokey > chain->bref.key)
861                                 lokey = chain->bref.key;
862                         adv = 0;
863                 }
864                 if (adv == 0)
865                         continue;
866
867                 /*
868                  * Advance element if possible, advanced element may be NULL.
869                  */
870                 if (chain)
871                         hammer2_chain_drop_unhold(chain);
872
873                 fifo = &xop->collect[i];
874                 if (fifo->ri != fifo->wi) {
875                         cpu_lfence();
876                         chain = fifo->array[fifo->ri & HAMMER2_XOPFIFO_MASK];
877                         error = fifo->errors[fifo->ri & HAMMER2_XOPFIFO_MASK];
878                         ++fifo->ri;
879                         xop->cluster.array[i].chain = chain;
880                         xop->cluster.array[i].error = error;
881                         if (chain == NULL) {
882                                 /* XXX */
883                                 xop->cluster.array[i].flags |=
884                                                         HAMMER2_CITEM_NULL;
885                         }
886                         if (fifo->wi - fifo->ri <= HAMMER2_XOPFIFO / 2) {
887                                 if (fifo->flags & HAMMER2_XOP_FIFO_STALL) {
888                                         atomic_clear_int(&fifo->flags,
889                                                     HAMMER2_XOP_FIFO_STALL);
890                                         wakeup(xop);
891                                         lwkt_yield();
892                                 }
893                         }
894                         --i;            /* loop on same index */
895                 } else {
896                         /*
897                          * Retain CITEM_NULL flag.  If set just repeat EOF.
898                          * If not, the NULL,0 combination indicates an
899                          * operation in-progress.
900                          */
901                         xop->cluster.array[i].chain = NULL;
902                         /* retain any CITEM_NULL setting */
903                 }
904         }
905
906         /*
907          * Determine whether the lowest collected key meets clustering
908          * requirements.  Returns:
909          *
910          * 0             - key valid, cluster can be returned.
911          *
912          * ENOENT        - normal end of scan, return ENOENT.
913          *
914          * ESRCH         - sufficient elements collected, quorum agreement
915          *                 that lokey is not a valid element and should be
916          *                 skipped.
917          *
918          * EDEADLK       - sufficient elements collected, no quorum agreement
919          *                 (and no agreement possible).  In this situation a
920          *                 repair is needed, for now we loop.
921          *
922          * EINPROGRESS   - insufficient elements collected to resolve, wait
923          *                 for event and loop.
924          */
925         if ((flags & HAMMER2_XOP_COLLECT_WAITALL) &&
926             (mask & HAMMER2_XOPMASK_ALLDONE) != HAMMER2_XOPMASK_VOP) {
927                 error = HAMMER2_ERROR_EINPROGRESS;
928         } else {
929                 error = hammer2_cluster_check(&xop->cluster, lokey, keynull);
930         }
931         if (error == HAMMER2_ERROR_EINPROGRESS) {
932                 if (flags & HAMMER2_XOP_COLLECT_NOWAIT)
933                         goto done;
934                 tsleep_interlock(xop, 0);
935                 if (atomic_cmpset_64(&xop->run_mask,
936                                      mask, mask | HAMMER2_XOPMASK_WAIT)) {
937                         tsleep(xop, PINTERLOCKED, "h2coll", hz*60);
938                 }
939                 goto loop;
940         }
941         if (error == HAMMER2_ERROR_ESRCH) {
942                 if (lokey != HAMMER2_KEY_MAX) {
943                         xop->collect_key = lokey + 1;
944                         goto loop;
945                 }
946                 error = HAMMER2_ERROR_ENOENT;
947         }
948         if (error == HAMMER2_ERROR_EDEADLK) {
949                 kprintf("hammer2: no quorum possible lokey %016jx\n",
950                         lokey);
951                 if (lokey != HAMMER2_KEY_MAX) {
952                         xop->collect_key = lokey + 1;
953                         goto loop;
954                 }
955                 error = HAMMER2_ERROR_ENOENT;
956         }
957         if (lokey == HAMMER2_KEY_MAX)
958                 xop->collect_key = lokey;
959         else
960                 xop->collect_key = lokey + 1;
961 done:
962         return error;
963 }
964
965 /*
966  * N x M processing threads are available to handle XOPs, N per cluster
967  * index x M cluster nodes.
968  *
969  * Locate and return the next runnable xop, or NULL if no xops are
970  * present or none of the xops are currently runnable (for various reasons).
971  * The xop is left on the queue and serves to block other dependent xops
972  * from being run.
973  *
974  * Dependent xops will not be returned.
975  *
976  * Sets HAMMER2_XOP_FIFO_RUN on the returned xop or returns NULL.
977  *
978  * NOTE! Xops run concurrently for each cluster index.
979  */
980 #define XOP_HASH_SIZE   16
981 #define XOP_HASH_MASK   (XOP_HASH_SIZE - 1)
982
983 static __inline
984 int
985 xop_testhash(hammer2_thread_t *thr, hammer2_inode_t *ip, uint32_t *hash)
986 {
987         uint32_t mask;
988         int hv;
989
990         hv = (int)((uintptr_t)ip + (uintptr_t)thr) / sizeof(hammer2_inode_t);
991         mask = 1U << (hv & 31);
992         hv >>= 5;
993
994         return ((int)(hash[hv & XOP_HASH_MASK] & mask));
995 }
996
997 static __inline
998 void
999 xop_sethash(hammer2_thread_t *thr, hammer2_inode_t *ip, uint32_t *hash)
1000 {
1001         uint32_t mask;
1002         int hv;
1003
1004         hv = (int)((uintptr_t)ip + (uintptr_t)thr) / sizeof(hammer2_inode_t);
1005         mask = 1U << (hv & 31);
1006         hv >>= 5;
1007
1008         hash[hv & XOP_HASH_MASK] |= mask;
1009 }
1010
1011 static
1012 hammer2_xop_head_t *
1013 hammer2_xop_next(hammer2_thread_t *thr)
1014 {
1015         hammer2_pfs_t *pmp = thr->pmp;
1016         int clindex = thr->clindex;
1017         uint32_t hash[XOP_HASH_SIZE] = { 0 };
1018         hammer2_xop_head_t *xop;
1019
1020         hammer2_spin_ex(&pmp->xop_spin);
1021         TAILQ_FOREACH(xop, &thr->xopq, collect[clindex].entry) {
1022                 /*
1023                  * Check dependency
1024                  */
1025                 if (xop_testhash(thr, xop->ip1, hash) ||
1026                     (xop->ip2 && xop_testhash(thr, xop->ip2, hash)) ||
1027                     (xop->ip3 && xop_testhash(thr, xop->ip3, hash))) {
1028                         continue;
1029                 }
1030                 xop_sethash(thr, xop->ip1, hash);
1031                 if (xop->ip2)
1032                         xop_sethash(thr, xop->ip2, hash);
1033                 if (xop->ip3)
1034                         xop_sethash(thr, xop->ip3, hash);
1035
1036                 /*
1037                  * Check already running
1038                  */
1039                 if (xop->collect[clindex].flags & HAMMER2_XOP_FIFO_RUN)
1040                         continue;
1041
1042                 /*
1043                  * Found a good one, return it.
1044                  */
1045                 atomic_set_int(&xop->collect[clindex].flags,
1046                                HAMMER2_XOP_FIFO_RUN);
1047                 break;
1048         }
1049         hammer2_spin_unex(&pmp->xop_spin);
1050
1051         return xop;
1052 }
1053
1054 /*
1055  * Remove the completed XOP from the queue, clear HAMMER2_XOP_FIFO_RUN.
1056  *
1057  * NOTE! Xops run concurrently for each cluster index.
1058  */
1059 static
1060 void
1061 hammer2_xop_dequeue(hammer2_thread_t *thr, hammer2_xop_head_t *xop)
1062 {
1063         hammer2_pfs_t *pmp = thr->pmp;
1064         int clindex = thr->clindex;
1065
1066         hammer2_spin_ex(&pmp->xop_spin);
1067         TAILQ_REMOVE(&thr->xopq, xop, collect[clindex].entry);
1068         atomic_clear_int(&xop->collect[clindex].flags,
1069                          HAMMER2_XOP_FIFO_RUN);
1070         hammer2_spin_unex(&pmp->xop_spin);
1071         if (TAILQ_FIRST(&thr->xopq))
1072                 hammer2_thr_signal(thr, HAMMER2_THREAD_XOPQ);
1073 }
1074
1075 /*
1076  * Primary management thread for xops support.  Each node has several such
1077  * threads which replicate front-end operations on cluster nodes.
1078  *
1079  * XOPS thread node operations, allowing the function to focus on a single
1080  * node in the cluster after validating the operation with the cluster.
1081  * This is primarily what prevents dead or stalled nodes from stalling
1082  * the front-end.
1083  */
1084 void
1085 hammer2_primary_xops_thread(void *arg)
1086 {
1087         hammer2_thread_t *thr = arg;
1088         hammer2_pfs_t *pmp;
1089         hammer2_xop_head_t *xop;
1090         uint64_t mask;
1091         uint32_t flags;
1092         uint32_t nflags;
1093         hammer2_xop_desc_t *last_desc = NULL;
1094
1095         pmp = thr->pmp;
1096         /*xgrp = &pmp->xop_groups[thr->repidx]; not needed */
1097         mask = 1LLU << thr->clindex;
1098
1099         for (;;) {
1100                 flags = thr->flags;
1101
1102                 /*
1103                  * Handle stop request
1104                  */
1105                 if (flags & HAMMER2_THREAD_STOP)
1106                         break;
1107
1108                 /*
1109                  * Handle freeze request
1110                  */
1111                 if (flags & HAMMER2_THREAD_FREEZE) {
1112                         hammer2_thr_signal2(thr, HAMMER2_THREAD_FROZEN,
1113                                                  HAMMER2_THREAD_FREEZE);
1114                         continue;
1115                 }
1116
1117                 if (flags & HAMMER2_THREAD_UNFREEZE) {
1118                         hammer2_thr_signal2(thr, 0,
1119                                                  HAMMER2_THREAD_FROZEN |
1120                                                  HAMMER2_THREAD_UNFREEZE);
1121                         continue;
1122                 }
1123
1124                 /*
1125                  * Force idle if frozen until unfrozen or stopped.
1126                  */
1127                 if (flags & HAMMER2_THREAD_FROZEN) {
1128                         hammer2_thr_wait_any(thr,
1129                                              HAMMER2_THREAD_UNFREEZE |
1130                                              HAMMER2_THREAD_STOP,
1131                                              0);
1132                         continue;
1133                 }
1134
1135                 /*
1136                  * Reset state on REMASTER request
1137                  */
1138                 if (flags & HAMMER2_THREAD_REMASTER) {
1139                         hammer2_thr_signal2(thr, 0, HAMMER2_THREAD_REMASTER);
1140                         /* reset state here */
1141                         continue;
1142                 }
1143
1144                 /*
1145                  * Process requests.  Each request can be multi-queued.
1146                  *
1147                  * If we get behind and the frontend VOP is no longer active,
1148                  * we retire the request without processing it.  The callback
1149                  * may also abort processing if the frontend VOP becomes
1150                  * inactive.
1151                  */
1152                 if (flags & HAMMER2_THREAD_XOPQ) {
1153                         nflags = flags & ~HAMMER2_THREAD_XOPQ;
1154                         if (!atomic_cmpset_int(&thr->flags, flags, nflags))
1155                                 continue;
1156                         flags = nflags;
1157                         /* fall through */
1158                 }
1159                 while ((xop = hammer2_xop_next(thr)) != NULL) {
1160                         if (hammer2_xop_active(xop)) {
1161                                 last_desc = xop->desc;
1162                                 xop->desc->storage_func((hammer2_xop_t *)xop,
1163                                                         thr->scratch,
1164                                                         thr->clindex);
1165                                 hammer2_xop_dequeue(thr, xop);
1166                                 hammer2_xop_retire(xop, mask);
1167                         } else {
1168                                 last_desc = xop->desc;
1169                                 hammer2_xop_feed(xop, NULL, thr->clindex,
1170                                                  ECONNABORTED);
1171                                 hammer2_xop_dequeue(thr, xop);
1172                                 hammer2_xop_retire(xop, mask);
1173                         }
1174                 }
1175
1176                 /*
1177                  * Wait for event, interlock using THREAD_WAITING and
1178                  * THREAD_SIGNAL.
1179                  *
1180                  * For robustness poll on a 30-second interval, but nominally
1181                  * expect to be woken up.
1182                  */
1183                 nflags = flags | HAMMER2_THREAD_WAITING;
1184
1185                 tsleep_interlock(&thr->flags, 0);
1186                 if (atomic_cmpset_int(&thr->flags, flags, nflags)) {
1187                         tsleep(&thr->flags, PINTERLOCKED, "h2idle", hz*30);
1188                 }
1189         }
1190
1191 #if 0
1192         /*
1193          * Cleanup / termination
1194          */
1195         while ((xop = TAILQ_FIRST(&thr->xopq)) != NULL) {
1196                 kprintf("hammer2_thread: aborting xop %s\n", xop->desc->id);
1197                 TAILQ_REMOVE(&thr->xopq, xop,
1198                              collect[thr->clindex].entry);
1199                 hammer2_xop_retire(xop, mask);
1200         }
1201 #endif
1202         thr->td = NULL;
1203         hammer2_thr_signal(thr, HAMMER2_THREAD_STOPPED);
1204         /* thr structure can go invalid after this point */
1205 }