1 // SPDX-License-Identifier: GPL-2.0 2 3 #include "bcachefs.h" 4 #include "btree_locking.h" 5 #include "btree_update.h" 6 #include "btree_update_interior.h" 7 #include "btree_write_buffer.h" 8 #include "error.h" 9 #include "journal.h" 10 #include "journal_io.h" 11 #include "journal_reclaim.h" 12 13 #include <linux/prefetch.h> 14 15 static int bch2_btree_write_buffer_journal_flush(struct journal *, 16 struct journal_entry_pin *, u64); 17 18 static int bch2_journal_keys_to_write_buffer(struct bch_fs *, struct journal_buf *); 19 20 static inline bool __wb_key_ref_cmp(const struct wb_key_ref *l, const struct wb_key_ref *r) 21 { 22 return (cmp_int(l->hi, r->hi) ?: 23 cmp_int(l->mi, r->mi) ?: 24 cmp_int(l->lo, r->lo)) >= 0; 25 } 26 27 static inline bool wb_key_ref_cmp(const struct wb_key_ref *l, const struct wb_key_ref *r) 28 { 29 #ifdef CONFIG_X86_64 30 int cmp; 31 32 asm("mov (%[l]), %%rax;" 33 "sub (%[r]), %%rax;" 34 "mov 8(%[l]), %%rax;" 35 "sbb 8(%[r]), %%rax;" 36 "mov 16(%[l]), %%rax;" 37 "sbb 16(%[r]), %%rax;" 38 : "=@ccae" (cmp) 39 : [l] "r" (l), [r] "r" (r) 40 : "rax", "cc"); 41 42 EBUG_ON(cmp != __wb_key_ref_cmp(l, r)); 43 return cmp; 44 #else 45 return __wb_key_ref_cmp(l, r); 46 #endif 47 } 48 49 /* Compare excluding idx, the low 24 bits: */ 50 static inline bool wb_key_eq(const void *_l, const void *_r) 51 { 52 const struct wb_key_ref *l = _l; 53 const struct wb_key_ref *r = _r; 54 55 return !((l->hi ^ r->hi)| 56 (l->mi ^ r->mi)| 57 ((l->lo >> 24) ^ (r->lo >> 24))); 58 } 59 60 static noinline void wb_sort(struct wb_key_ref *base, size_t num) 61 { 62 size_t n = num, a = num / 2; 63 64 if (!a) /* num < 2 || size == 0 */ 65 return; 66 67 for (;;) { 68 size_t b, c, d; 69 70 if (a) /* Building heap: sift down --a */ 71 --a; 72 else if (--n) /* Sorting: Extract root to --n */ 73 swap(base[0], base[n]); 74 else /* Sort complete */ 75 break; 76 77 /* 78 * Sift element at "a" down into heap. This is the 79 * "bottom-up" variant, which significantly reduces 80 * calls to cmp_func(): we find the sift-down path all 81 * the way to the leaves (one compare per level), then 82 * backtrack to find where to insert the target element. 83 * 84 * Because elements tend to sift down close to the leaves, 85 * this uses fewer compares than doing two per level 86 * on the way down. (A bit more than half as many on 87 * average, 3/4 worst-case.) 88 */ 89 for (b = a; c = 2*b + 1, (d = c + 1) < n;) 90 b = wb_key_ref_cmp(base + c, base + d) ? c : d; 91 if (d == n) /* Special case last leaf with no sibling */ 92 b = c; 93 94 /* Now backtrack from "b" to the correct location for "a" */ 95 while (b != a && wb_key_ref_cmp(base + a, base + b)) 96 b = (b - 1) / 2; 97 c = b; /* Where "a" belongs */ 98 while (b != a) { /* Shift it into place */ 99 b = (b - 1) / 2; 100 swap(base[b], base[c]); 101 } 102 } 103 } 104 105 static noinline int wb_flush_one_slowpath(struct btree_trans *trans, 106 struct btree_iter *iter, 107 struct btree_write_buffered_key *wb) 108 { 109 struct btree_path *path = btree_iter_path(trans, iter); 110 111 bch2_btree_node_unlock_write(trans, path, path->l[0].b); 112 113 trans->journal_res.seq = wb->journal_seq; 114 115 return bch2_trans_update(trans, iter, &wb->k, 116 BTREE_UPDATE_INTERNAL_SNAPSHOT_NODE) ?: 117 bch2_trans_commit(trans, NULL, NULL, 118 BCH_TRANS_COMMIT_no_enospc| 119 BCH_TRANS_COMMIT_no_check_rw| 120 BCH_TRANS_COMMIT_no_journal_res| 121 BCH_TRANS_COMMIT_journal_reclaim); 122 } 123 124 static inline int wb_flush_one(struct btree_trans *trans, struct btree_iter *iter, 125 struct btree_write_buffered_key *wb, 126 bool *write_locked, size_t *fast) 127 { 128 struct bch_fs *c = trans->c; 129 struct btree_path *path; 130 int ret; 131 132 EBUG_ON(!wb->journal_seq); 133 EBUG_ON(!c->btree_write_buffer.flushing.pin.seq); 134 EBUG_ON(c->btree_write_buffer.flushing.pin.seq > wb->journal_seq); 135 136 ret = bch2_btree_iter_traverse(iter); 137 if (ret) 138 return ret; 139 140 /* 141 * We can't clone a path that has write locks: unshare it now, before 142 * set_pos and traverse(): 143 */ 144 if (btree_iter_path(trans, iter)->ref > 1) 145 iter->path = __bch2_btree_path_make_mut(trans, iter->path, true, _THIS_IP_); 146 147 path = btree_iter_path(trans, iter); 148 149 if (!*write_locked) { 150 ret = bch2_btree_node_lock_write(trans, path, &path->l[0].b->c); 151 if (ret) 152 return ret; 153 154 bch2_btree_node_prep_for_write(trans, path, path->l[0].b); 155 *write_locked = true; 156 } 157 158 if (unlikely(!bch2_btree_node_insert_fits(c, path->l[0].b, wb->k.k.u64s))) { 159 *write_locked = false; 160 return wb_flush_one_slowpath(trans, iter, wb); 161 } 162 163 bch2_btree_insert_key_leaf(trans, path, &wb->k, wb->journal_seq); 164 (*fast)++; 165 return 0; 166 } 167 168 /* 169 * Update a btree with a write buffered key using the journal seq of the 170 * original write buffer insert. 171 * 172 * It is not safe to rejournal the key once it has been inserted into the write 173 * buffer because that may break recovery ordering. For example, the key may 174 * have already been modified in the active write buffer in a seq that comes 175 * before the current transaction. If we were to journal this key again and 176 * crash, recovery would process updates in the wrong order. 177 */ 178 static int 179 btree_write_buffered_insert(struct btree_trans *trans, 180 struct btree_write_buffered_key *wb) 181 { 182 struct btree_iter iter; 183 int ret; 184 185 bch2_trans_iter_init(trans, &iter, wb->btree, bkey_start_pos(&wb->k.k), 186 BTREE_ITER_CACHED|BTREE_ITER_INTENT); 187 188 trans->journal_res.seq = wb->journal_seq; 189 190 ret = bch2_btree_iter_traverse(&iter) ?: 191 bch2_trans_update(trans, &iter, &wb->k, 192 BTREE_UPDATE_INTERNAL_SNAPSHOT_NODE); 193 bch2_trans_iter_exit(trans, &iter); 194 return ret; 195 } 196 197 static void move_keys_from_inc_to_flushing(struct btree_write_buffer *wb) 198 { 199 struct bch_fs *c = container_of(wb, struct bch_fs, btree_write_buffer); 200 struct journal *j = &c->journal; 201 202 if (!wb->inc.keys.nr) 203 return; 204 205 bch2_journal_pin_add(j, wb->inc.keys.data[0].journal_seq, &wb->flushing.pin, 206 bch2_btree_write_buffer_journal_flush); 207 208 darray_resize(&wb->flushing.keys, min_t(size_t, 1U << 20, wb->flushing.keys.nr + wb->inc.keys.nr)); 209 darray_resize(&wb->sorted, wb->flushing.keys.size); 210 211 if (!wb->flushing.keys.nr && wb->sorted.size >= wb->inc.keys.nr) { 212 swap(wb->flushing.keys, wb->inc.keys); 213 goto out; 214 } 215 216 size_t nr = min(darray_room(wb->flushing.keys), 217 wb->sorted.size - wb->flushing.keys.nr); 218 nr = min(nr, wb->inc.keys.nr); 219 220 memcpy(&darray_top(wb->flushing.keys), 221 wb->inc.keys.data, 222 sizeof(wb->inc.keys.data[0]) * nr); 223 224 memmove(wb->inc.keys.data, 225 wb->inc.keys.data + nr, 226 sizeof(wb->inc.keys.data[0]) * (wb->inc.keys.nr - nr)); 227 228 wb->flushing.keys.nr += nr; 229 wb->inc.keys.nr -= nr; 230 out: 231 if (!wb->inc.keys.nr) 232 bch2_journal_pin_drop(j, &wb->inc.pin); 233 else 234 bch2_journal_pin_update(j, wb->inc.keys.data[0].journal_seq, &wb->inc.pin, 235 bch2_btree_write_buffer_journal_flush); 236 237 if (j->watermark) { 238 spin_lock(&j->lock); 239 bch2_journal_set_watermark(j); 240 spin_unlock(&j->lock); 241 } 242 243 BUG_ON(wb->sorted.size < wb->flushing.keys.nr); 244 } 245 246 static int bch2_btree_write_buffer_flush_locked(struct btree_trans *trans) 247 { 248 struct bch_fs *c = trans->c; 249 struct journal *j = &c->journal; 250 struct btree_write_buffer *wb = &c->btree_write_buffer; 251 struct btree_iter iter = { NULL }; 252 size_t skipped = 0, fast = 0, slowpath = 0; 253 bool write_locked = false; 254 int ret = 0; 255 256 bch2_trans_unlock(trans); 257 bch2_trans_begin(trans); 258 259 mutex_lock(&wb->inc.lock); 260 move_keys_from_inc_to_flushing(wb); 261 mutex_unlock(&wb->inc.lock); 262 263 for (size_t i = 0; i < wb->flushing.keys.nr; i++) { 264 wb->sorted.data[i].idx = i; 265 wb->sorted.data[i].btree = wb->flushing.keys.data[i].btree; 266 memcpy(&wb->sorted.data[i].pos, &wb->flushing.keys.data[i].k.k.p, sizeof(struct bpos)); 267 } 268 wb->sorted.nr = wb->flushing.keys.nr; 269 270 /* 271 * We first sort so that we can detect and skip redundant updates, and 272 * then we attempt to flush in sorted btree order, as this is most 273 * efficient. 274 * 275 * However, since we're not flushing in the order they appear in the 276 * journal we won't be able to drop our journal pin until everything is 277 * flushed - which means this could deadlock the journal if we weren't 278 * passing BCH_TRANS_COMMIT_journal_reclaim. This causes the update to fail 279 * if it would block taking a journal reservation. 280 * 281 * If that happens, simply skip the key so we can optimistically insert 282 * as many keys as possible in the fast path. 283 */ 284 wb_sort(wb->sorted.data, wb->sorted.nr); 285 286 darray_for_each(wb->sorted, i) { 287 struct btree_write_buffered_key *k = &wb->flushing.keys.data[i->idx]; 288 289 for (struct wb_key_ref *n = i + 1; n < min(i + 4, &darray_top(wb->sorted)); n++) 290 prefetch(&wb->flushing.keys.data[n->idx]); 291 292 BUG_ON(!k->journal_seq); 293 294 if (i + 1 < &darray_top(wb->sorted) && 295 wb_key_eq(i, i + 1)) { 296 struct btree_write_buffered_key *n = &wb->flushing.keys.data[i[1].idx]; 297 298 skipped++; 299 n->journal_seq = min_t(u64, n->journal_seq, k->journal_seq); 300 k->journal_seq = 0; 301 continue; 302 } 303 304 if (write_locked) { 305 struct btree_path *path = btree_iter_path(trans, &iter); 306 307 if (path->btree_id != i->btree || 308 bpos_gt(k->k.k.p, path->l[0].b->key.k.p)) { 309 bch2_btree_node_unlock_write(trans, path, path->l[0].b); 310 write_locked = false; 311 } 312 } 313 314 if (!iter.path || iter.btree_id != k->btree) { 315 bch2_trans_iter_exit(trans, &iter); 316 bch2_trans_iter_init(trans, &iter, k->btree, k->k.k.p, 317 BTREE_ITER_INTENT|BTREE_ITER_ALL_SNAPSHOTS); 318 } 319 320 bch2_btree_iter_set_pos(&iter, k->k.k.p); 321 btree_iter_path(trans, &iter)->preserve = false; 322 323 do { 324 if (race_fault()) { 325 ret = -BCH_ERR_journal_reclaim_would_deadlock; 326 break; 327 } 328 329 ret = wb_flush_one(trans, &iter, k, &write_locked, &fast); 330 if (!write_locked) 331 bch2_trans_begin(trans); 332 } while (bch2_err_matches(ret, BCH_ERR_transaction_restart)); 333 334 if (!ret) { 335 k->journal_seq = 0; 336 } else if (ret == -BCH_ERR_journal_reclaim_would_deadlock) { 337 slowpath++; 338 ret = 0; 339 } else 340 break; 341 } 342 343 if (write_locked) { 344 struct btree_path *path = btree_iter_path(trans, &iter); 345 bch2_btree_node_unlock_write(trans, path, path->l[0].b); 346 } 347 bch2_trans_iter_exit(trans, &iter); 348 349 if (ret) 350 goto err; 351 352 if (slowpath) { 353 /* 354 * Flush in the order they were present in the journal, so that 355 * we can release journal pins: 356 * The fastpath zapped the seq of keys that were successfully flushed so 357 * we can skip those here. 358 */ 359 trace_and_count(c, write_buffer_flush_slowpath, trans, slowpath, wb->flushing.keys.nr); 360 361 darray_for_each(wb->flushing.keys, i) { 362 if (!i->journal_seq) 363 continue; 364 365 bch2_journal_pin_update(j, i->journal_seq, &wb->flushing.pin, 366 bch2_btree_write_buffer_journal_flush); 367 368 bch2_trans_begin(trans); 369 370 ret = commit_do(trans, NULL, NULL, 371 BCH_WATERMARK_reclaim| 372 BCH_TRANS_COMMIT_no_check_rw| 373 BCH_TRANS_COMMIT_no_enospc| 374 BCH_TRANS_COMMIT_no_journal_res| 375 BCH_TRANS_COMMIT_journal_reclaim, 376 btree_write_buffered_insert(trans, i)); 377 if (ret) 378 goto err; 379 } 380 } 381 err: 382 bch2_fs_fatal_err_on(ret, c, "%s: insert error %s", __func__, bch2_err_str(ret)); 383 trace_write_buffer_flush(trans, wb->flushing.keys.nr, skipped, fast, 0); 384 bch2_journal_pin_drop(j, &wb->flushing.pin); 385 wb->flushing.keys.nr = 0; 386 return ret; 387 } 388 389 static int fetch_wb_keys_from_journal(struct bch_fs *c, u64 seq) 390 { 391 struct journal *j = &c->journal; 392 struct journal_buf *buf; 393 int ret = 0; 394 395 while (!ret && (buf = bch2_next_write_buffer_flush_journal_buf(j, seq))) { 396 ret = bch2_journal_keys_to_write_buffer(c, buf); 397 mutex_unlock(&j->buf_lock); 398 } 399 400 return ret; 401 } 402 403 static int btree_write_buffer_flush_seq(struct btree_trans *trans, u64 seq) 404 { 405 struct bch_fs *c = trans->c; 406 struct btree_write_buffer *wb = &c->btree_write_buffer; 407 int ret = 0, fetch_from_journal_err; 408 409 do { 410 bch2_trans_unlock(trans); 411 412 fetch_from_journal_err = fetch_wb_keys_from_journal(c, seq); 413 414 /* 415 * On memory allocation failure, bch2_btree_write_buffer_flush_locked() 416 * is not guaranteed to empty wb->inc: 417 */ 418 mutex_lock(&wb->flushing.lock); 419 ret = bch2_btree_write_buffer_flush_locked(trans); 420 mutex_unlock(&wb->flushing.lock); 421 } while (!ret && 422 (fetch_from_journal_err || 423 (wb->inc.pin.seq && wb->inc.pin.seq <= seq) || 424 (wb->flushing.pin.seq && wb->flushing.pin.seq <= seq))); 425 426 return ret; 427 } 428 429 static int bch2_btree_write_buffer_journal_flush(struct journal *j, 430 struct journal_entry_pin *_pin, u64 seq) 431 { 432 struct bch_fs *c = container_of(j, struct bch_fs, journal); 433 434 return bch2_trans_run(c, btree_write_buffer_flush_seq(trans, seq)); 435 } 436 437 int bch2_btree_write_buffer_flush_sync(struct btree_trans *trans) 438 { 439 struct bch_fs *c = trans->c; 440 441 trace_and_count(c, write_buffer_flush_sync, trans, _RET_IP_); 442 443 return btree_write_buffer_flush_seq(trans, journal_cur_seq(&c->journal)); 444 } 445 446 int bch2_btree_write_buffer_flush_nocheck_rw(struct btree_trans *trans) 447 { 448 struct bch_fs *c = trans->c; 449 struct btree_write_buffer *wb = &c->btree_write_buffer; 450 int ret = 0; 451 452 if (mutex_trylock(&wb->flushing.lock)) { 453 ret = bch2_btree_write_buffer_flush_locked(trans); 454 mutex_unlock(&wb->flushing.lock); 455 } 456 457 return ret; 458 } 459 460 int bch2_btree_write_buffer_tryflush(struct btree_trans *trans) 461 { 462 struct bch_fs *c = trans->c; 463 464 if (!bch2_write_ref_tryget(c, BCH_WRITE_REF_btree_write_buffer)) 465 return -BCH_ERR_erofs_no_writes; 466 467 int ret = bch2_btree_write_buffer_flush_nocheck_rw(trans); 468 bch2_write_ref_put(c, BCH_WRITE_REF_btree_write_buffer); 469 return ret; 470 } 471 472 static void bch2_btree_write_buffer_flush_work(struct work_struct *work) 473 { 474 struct bch_fs *c = container_of(work, struct bch_fs, btree_write_buffer.flush_work); 475 struct btree_write_buffer *wb = &c->btree_write_buffer; 476 int ret; 477 478 mutex_lock(&wb->flushing.lock); 479 do { 480 ret = bch2_trans_run(c, bch2_btree_write_buffer_flush_locked(trans)); 481 } while (!ret && bch2_btree_write_buffer_should_flush(c)); 482 mutex_unlock(&wb->flushing.lock); 483 484 bch2_write_ref_put(c, BCH_WRITE_REF_btree_write_buffer); 485 } 486 487 int bch2_journal_key_to_wb_slowpath(struct bch_fs *c, 488 struct journal_keys_to_wb *dst, 489 enum btree_id btree, struct bkey_i *k) 490 { 491 struct btree_write_buffer *wb = &c->btree_write_buffer; 492 int ret; 493 retry: 494 ret = darray_make_room_gfp(&dst->wb->keys, 1, GFP_KERNEL); 495 if (!ret && dst->wb == &wb->flushing) 496 ret = darray_resize(&wb->sorted, wb->flushing.keys.size); 497 498 if (unlikely(ret)) { 499 if (dst->wb == &c->btree_write_buffer.flushing) { 500 mutex_unlock(&dst->wb->lock); 501 dst->wb = &c->btree_write_buffer.inc; 502 bch2_journal_pin_add(&c->journal, dst->seq, &dst->wb->pin, 503 bch2_btree_write_buffer_journal_flush); 504 goto retry; 505 } 506 507 return ret; 508 } 509 510 dst->room = darray_room(dst->wb->keys); 511 if (dst->wb == &wb->flushing) 512 dst->room = min(dst->room, wb->sorted.size - wb->flushing.keys.nr); 513 BUG_ON(!dst->room); 514 BUG_ON(!dst->seq); 515 516 struct btree_write_buffered_key *wb_k = &darray_top(dst->wb->keys); 517 wb_k->journal_seq = dst->seq; 518 wb_k->btree = btree; 519 bkey_copy(&wb_k->k, k); 520 dst->wb->keys.nr++; 521 dst->room--; 522 return 0; 523 } 524 525 void bch2_journal_keys_to_write_buffer_start(struct bch_fs *c, struct journal_keys_to_wb *dst, u64 seq) 526 { 527 struct btree_write_buffer *wb = &c->btree_write_buffer; 528 529 if (mutex_trylock(&wb->flushing.lock)) { 530 mutex_lock(&wb->inc.lock); 531 move_keys_from_inc_to_flushing(wb); 532 533 /* 534 * Attempt to skip wb->inc, and add keys directly to 535 * wb->flushing, saving us a copy later: 536 */ 537 538 if (!wb->inc.keys.nr) { 539 dst->wb = &wb->flushing; 540 } else { 541 mutex_unlock(&wb->flushing.lock); 542 dst->wb = &wb->inc; 543 } 544 } else { 545 mutex_lock(&wb->inc.lock); 546 dst->wb = &wb->inc; 547 } 548 549 dst->room = darray_room(dst->wb->keys); 550 if (dst->wb == &wb->flushing) 551 dst->room = min(dst->room, wb->sorted.size - wb->flushing.keys.nr); 552 dst->seq = seq; 553 554 bch2_journal_pin_add(&c->journal, seq, &dst->wb->pin, 555 bch2_btree_write_buffer_journal_flush); 556 } 557 558 void bch2_journal_keys_to_write_buffer_end(struct bch_fs *c, struct journal_keys_to_wb *dst) 559 { 560 struct btree_write_buffer *wb = &c->btree_write_buffer; 561 562 if (!dst->wb->keys.nr) 563 bch2_journal_pin_drop(&c->journal, &dst->wb->pin); 564 565 if (bch2_btree_write_buffer_should_flush(c) && 566 __bch2_write_ref_tryget(c, BCH_WRITE_REF_btree_write_buffer) && 567 !queue_work(system_unbound_wq, &c->btree_write_buffer.flush_work)) 568 bch2_write_ref_put(c, BCH_WRITE_REF_btree_write_buffer); 569 570 if (dst->wb == &wb->flushing) 571 mutex_unlock(&wb->flushing.lock); 572 mutex_unlock(&wb->inc.lock); 573 } 574 575 static int bch2_journal_keys_to_write_buffer(struct bch_fs *c, struct journal_buf *buf) 576 { 577 struct journal_keys_to_wb dst; 578 struct jset_entry *entry; 579 struct bkey_i *k; 580 int ret = 0; 581 582 bch2_journal_keys_to_write_buffer_start(c, &dst, le64_to_cpu(buf->data->seq)); 583 584 for_each_jset_entry_type(entry, buf->data, BCH_JSET_ENTRY_write_buffer_keys) { 585 jset_entry_for_each_key(entry, k) { 586 ret = bch2_journal_key_to_wb(c, &dst, entry->btree_id, k); 587 if (ret) 588 goto out; 589 } 590 591 entry->type = BCH_JSET_ENTRY_btree_keys; 592 } 593 594 buf->need_flush_to_write_buffer = false; 595 out: 596 bch2_journal_keys_to_write_buffer_end(c, &dst); 597 return ret; 598 } 599 600 static int wb_keys_resize(struct btree_write_buffer_keys *wb, size_t new_size) 601 { 602 if (wb->keys.size >= new_size) 603 return 0; 604 605 if (!mutex_trylock(&wb->lock)) 606 return -EINTR; 607 608 int ret = darray_resize(&wb->keys, new_size); 609 mutex_unlock(&wb->lock); 610 return ret; 611 } 612 613 int bch2_btree_write_buffer_resize(struct bch_fs *c, size_t new_size) 614 { 615 struct btree_write_buffer *wb = &c->btree_write_buffer; 616 617 return wb_keys_resize(&wb->flushing, new_size) ?: 618 wb_keys_resize(&wb->inc, new_size); 619 } 620 621 void bch2_fs_btree_write_buffer_exit(struct bch_fs *c) 622 { 623 struct btree_write_buffer *wb = &c->btree_write_buffer; 624 625 BUG_ON((wb->inc.keys.nr || wb->flushing.keys.nr) && 626 !bch2_journal_error(&c->journal)); 627 628 darray_exit(&wb->sorted); 629 darray_exit(&wb->flushing.keys); 630 darray_exit(&wb->inc.keys); 631 } 632 633 int bch2_fs_btree_write_buffer_init(struct bch_fs *c) 634 { 635 struct btree_write_buffer *wb = &c->btree_write_buffer; 636 637 mutex_init(&wb->inc.lock); 638 mutex_init(&wb->flushing.lock); 639 INIT_WORK(&wb->flush_work, bch2_btree_write_buffer_flush_work); 640 641 /* Will be resized by journal as needed: */ 642 unsigned initial_size = 1 << 16; 643 644 return darray_make_room(&wb->inc.keys, initial_size) ?: 645 darray_make_room(&wb->flushing.keys, initial_size) ?: 646 darray_make_room(&wb->sorted, initial_size); 647 } 648