ceph/src/osdc/Journaler.cc

   1 // -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
   2 // vim: ts=8 sw=2 smarttab
   3 /*
   4  * Ceph - scalable distributed file system
   5  *
   6  * Copyright (C) 2004-2006 Sage Weil <sage@newdream.net>
   7  *
   8  * This is free software; you can redistribute it and/or
   9  * modify it under the terms of the GNU Lesser General Public
  10  * License version 2.1, as published by the Free Software
  11  * Foundation.  See file COPYING.
  12  *
  13  */
  14
  15 #include "common/perf_counters.h"
  16 #include "common/dout.h"
  17 #include "include/Context.h"
  18 #include "msg/Messenger.h"
  19 #include "osdc/Journaler.h"
  20 #include "common/errno.h"
  21 #include "include/assert.h"
  22 #include "common/Finisher.h"
  23
  24 #define dout_subsys ceph_subsys_journaler
  25 #undef dout_prefix
  26 #define dout_prefix *_dout << objecter->messenger->get_myname() \
  27   << ".journaler." << name << (readonly ? "(ro) ":"(rw) ")
  28
  29 using std::chrono::seconds;
  30
  31
  32 class Journaler::C_DelayFlush : public Context {
  33   Journaler *journaler;
  34   public:
  35   C_DelayFlush(Journaler *j) : journaler(j) {}
  36   void finish(int r) override {
  37     journaler->_do_delayed_flush();
  38   }
  39 };
  40
  41 void Journaler::set_readonly()
  42 {
  43   lock_guard l(lock);
  44
  45   ldout(cct, 1) << "set_readonly" << dendl;
  46   readonly = true;
  47 }
  48
  49 void Journaler::set_writeable()
  50 {
  51   lock_guard l(lock);
  52
  53   ldout(cct, 1) << "set_writeable" << dendl;
  54   readonly = false;
  55 }
  56
  57 void Journaler::create(file_layout_t *l, stream_format_t const sf)
  58 {
  59   lock_guard lk(lock);
  60
  61   assert(!readonly);
  62   state = STATE_ACTIVE;
  63
  64   stream_format = sf;
  65   journal_stream.set_format(sf);
  66   _set_layout(l);
  67
  68   prezeroing_pos = prezero_pos = write_pos = flush_pos =
  69     safe_pos = read_pos = requested_pos = received_pos =
  70     expire_pos = trimming_pos = trimmed_pos =
  71     next_safe_pos = layout.get_period();
  72
  73   ldout(cct, 1) << "created blank journal at inode 0x" << std::hex << ino
  74                 << std::dec << ", format=" << stream_format << dendl;
  75 }
  76
  77 void Journaler::set_layout(file_layout_t const *l)
  78 {
  79     lock_guard lk(lock);
  80     _set_layout(l);
  81 }
  82
  83 void Journaler::_set_layout(file_layout_t const *l)
  84 {
  85   layout = *l;
  86
  87   assert(layout.pool_id == pg_pool);
  88   last_written.layout = layout;
  89   last_committed.layout = layout;
  90
  91   // prefetch intelligently.
  92   // (watch out, this is big if you use big objects or weird striping)
  93   uint64_t periods = cct->_conf->journaler_prefetch_periods;
  94   if (periods < 2)
  95     periods = 2;  // we need at least 2 periods to make progress.
  96   fetch_len = layout.get_period() * periods;
  97 }
  98
  99
 100 /***************** HEADER *******************/
 101
 102 ostream& operator<<(ostream &out, const Journaler::Header &h)
 103 {
 104   return out << "loghead(trim " << h.trimmed_pos
 105              << ", expire " << h.expire_pos
 106              << ", write " << h.write_pos
 107              << ", stream_format " << (int)(h.stream_format)
 108              << ")";
 109 }
 110
 111 class Journaler::C_ReadHead : public Context {
 112   Journaler *ls;
 113 public:
 114   bufferlist bl;
 115   explicit C_ReadHead(Journaler *l) : ls(l) {}
 116   void finish(int r) override {
 117     ls->_finish_read_head(r, bl);
 118   }
 119 };
 120
 121 class Journaler::C_RereadHead : public Context {
 122   Journaler *ls;
 123   Context *onfinish;
 124 public:
 125   bufferlist bl;
 126   C_RereadHead(Journaler *l, Context *onfinish_) : ls (l),
 127                                                    onfinish(onfinish_) {}
 128   void finish(int r) override {
 129     ls->_finish_reread_head(r, bl, onfinish);
 130   }
 131 };
 132
 133 class Journaler::C_ProbeEnd : public Context {
 134   Journaler *ls;
 135 public:
 136   uint64_t end;
 137   explicit C_ProbeEnd(Journaler *l) : ls(l), end(-1) {}
 138   void finish(int r) override {
 139     ls->_finish_probe_end(r, end);
 140   }
 141 };
 142
 143 class Journaler::C_ReProbe : public Context {
 144   Journaler *ls;
 145   C_OnFinisher *onfinish;
 146 public:
 147   uint64_t end;
 148   C_ReProbe(Journaler *l, C_OnFinisher *onfinish_) :
 149     ls(l), onfinish(onfinish_), end(0) {}
 150   void finish(int r) override {
 151     ls->_finish_reprobe(r, end, onfinish);
 152   }
 153 };
 154
 155 void Journaler::recover(Context *onread)
 156 {
 157   lock_guard l(lock);
 158   if (stopping) {
 159     onread->complete(-EAGAIN);
 160     return;
 161   }
 162
 163   ldout(cct, 1) << "recover start" << dendl;
 164   assert(state != STATE_ACTIVE);
 165   assert(readonly);
 166
 167   if (onread)
 168     waitfor_recover.push_back(wrap_finisher(onread));
 169
 170   if (state != STATE_UNDEF) {
 171     ldout(cct, 1) << "recover - already recovering" << dendl;
 172     return;
 173   }
 174
 175   ldout(cct, 1) << "read_head" << dendl;
 176   state = STATE_READHEAD;
 177   C_ReadHead *fin = new C_ReadHead(this);
 178   _read_head(fin, &fin->bl);
 179 }
 180
 181 void Journaler::_read_head(Context *on_finish, bufferlist *bl)
 182 {
 183   // lock is locked
 184   assert(state == STATE_READHEAD || state == STATE_REREADHEAD);
 185
 186   object_t oid = file_object_t(ino, 0);
 187   object_locator_t oloc(pg_pool);
 188   objecter->read_full(oid, oloc, CEPH_NOSNAP, bl, 0, wrap_finisher(on_finish));
 189 }
 190
 191 void Journaler::reread_head(Context *onfinish)
 192 {
 193   lock_guard l(lock);
 194   _reread_head(wrap_finisher(onfinish));
 195 }
 196
 197 /**
 198  * Re-read the head from disk, and set the write_pos, expire_pos, trimmed_pos
 199  * from the on-disk header. This switches the state to STATE_REREADHEAD for
 200  * the duration, and you shouldn't start a re-read while other operations are
 201  * in-flight, nor start other operations while a re-read is in progress.
 202  * Also, don't call this until the Journaler has finished its recovery and has
 203  * gone STATE_ACTIVE!
 204  */
 205 void Journaler::_reread_head(Context *onfinish)
 206 {
 207   ldout(cct, 10) << "reread_head" << dendl;
 208   assert(state == STATE_ACTIVE);
 209
 210   state = STATE_REREADHEAD;
 211   C_RereadHead *fin = new C_RereadHead(this, onfinish);
 212   _read_head(fin, &fin->bl);
 213 }
 214
 215 void Journaler::_finish_reread_head(int r, bufferlist& bl, Context *finish)
 216 {
 217   lock_guard l(lock);
 218
 219   //read on-disk header into
 220   assert(bl.length() || r < 0 );
 221
 222   // unpack header
 223   if (r == 0) {
 224     Header h;
 225     bufferlist::iterator p = bl.begin();
 226     try {
 227       ::decode(h, p);
 228     } catch (const buffer::error &e) {
 229       finish->complete(-EINVAL);
 230       return;
 231     }
 232     prezeroing_pos = prezero_pos = write_pos = flush_pos = safe_pos = next_safe_pos
 233       = h.write_pos;
 234     expire_pos = h.expire_pos;
 235     trimmed_pos = trimming_pos = h.trimmed_pos;
 236     init_headers(h);
 237     state = STATE_ACTIVE;
 238   }
 239
 240   finish->complete(r);
 241 }
 242
 243 void Journaler::_finish_read_head(int r, bufferlist& bl)
 244 {
 245   lock_guard l(lock);
 246
 247   assert(state == STATE_READHEAD);
 248
 249   if (r!=0) {
 250     ldout(cct, 0) << "error getting journal off disk" << dendl;
 251     list<Context*> ls;
 252     ls.swap(waitfor_recover);
 253     finish_contexts(cct, ls, r);
 254     return;
 255   }
 256
 257   if (bl.length() == 0) {
 258     ldout(cct, 1) << "_finish_read_head r=" << r
 259                   << " read 0 bytes, assuming empty log" << dendl;
 260     state = STATE_ACTIVE;
 261     list<Context*> ls;
 262     ls.swap(waitfor_recover);
 263     finish_contexts(cct, ls, 0);
 264     return;
 265   }
 266
 267   // unpack header
 268   bool corrupt = false;
 269   Header h;
 270   bufferlist::iterator p = bl.begin();
 271   try {
 272     ::decode(h, p);
 273
 274     if (h.magic != magic) {
 275       ldout(cct, 0) << "on disk magic '" << h.magic << "' != my magic '"
 276                     << magic << "'" << dendl;
 277       corrupt = true;
 278     } else if (h.write_pos < h.expire_pos || h.expire_pos < h.trimmed_pos) {
 279       ldout(cct, 0) << "Corrupt header (bad offsets): " << h << dendl;
 280       corrupt = true;
 281     }
 282   } catch (const buffer::error &e) {
 283     corrupt = true;
 284   }
 285
 286   if (corrupt) {
 287     list<Context*> ls;
 288     ls.swap(waitfor_recover);
 289     finish_contexts(cct, ls, -EINVAL);
 290     return;
 291   }
 292
 293   prezeroing_pos = prezero_pos = write_pos = flush_pos = safe_pos = next_safe_pos
 294     = h.write_pos;
 295   read_pos = requested_pos = received_pos = expire_pos = h.expire_pos;
 296   trimmed_pos = trimming_pos = h.trimmed_pos;
 297
 298   init_headers(h);
 299   _set_layout(&h.layout);
 300   stream_format = h.stream_format;
 301   journal_stream.set_format(h.stream_format);
 302
 303   ldout(cct, 1) << "_finish_read_head " << h
 304                 << ".  probing for end of log (from " << write_pos << ")..."
 305                 << dendl;
 306   C_ProbeEnd *fin = new C_ProbeEnd(this);
 307   state = STATE_PROBING;
 308   _probe(fin, &fin->end);
 309 }
 310
 311 void Journaler::_probe(Context *finish, uint64_t *end)
 312 {
 313   // lock is locked
 314   ldout(cct, 1) << "probing for end of the log" << dendl;
 315   assert(state == STATE_PROBING || state == STATE_REPROBING);
 316   // probe the log
 317   filer.probe(ino, &layout, CEPH_NOSNAP,
 318               write_pos, end, true, 0, wrap_finisher(finish));
 319 }
 320
 321 void Journaler::_reprobe(C_OnFinisher *finish)
 322 {
 323   ldout(cct, 10) << "reprobe" << dendl;
 324   assert(state == STATE_ACTIVE);
 325
 326   state = STATE_REPROBING;
 327   C_ReProbe *fin = new C_ReProbe(this, finish);
 328   _probe(fin, &fin->end);
 329 }
 330
 331
 332 void Journaler::_finish_reprobe(int r, uint64_t new_end,
 333                                 C_OnFinisher *onfinish)
 334 {
 335   lock_guard l(lock);
 336
 337   assert(new_end >= write_pos || r < 0);
 338   ldout(cct, 1) << "_finish_reprobe new_end = " << new_end
 339           << " (header had " << write_pos << ")."
 340           << dendl;
 341   prezeroing_pos = prezero_pos = write_pos = flush_pos = safe_pos = next_safe_pos = new_end;
 342   state = STATE_ACTIVE;
 343   onfinish->complete(r);
 344 }
 345
 346 void Journaler::_finish_probe_end(int r, uint64_t end)
 347 {
 348   lock_guard l(lock);
 349
 350   assert(state == STATE_PROBING);
 351   if (r < 0) { // error in probing
 352     goto out;
 353   }
 354   if (((int64_t)end) == -1) {
 355     end = write_pos;
 356     ldout(cct, 1) << "_finish_probe_end write_pos = " << end << " (header had "
 357                   << write_pos << "). log was empty. recovered." << dendl;
 358     ceph_abort(); // hrm.
 359   } else {
 360     assert(end >= write_pos);
 361     ldout(cct, 1) << "_finish_probe_end write_pos = " << end
 362                   << " (header had " << write_pos << "). recovered."
 363                   << dendl;
 364   }
 365
 366   state = STATE_ACTIVE;
 367
 368   prezeroing_pos = prezero_pos = write_pos = flush_pos = safe_pos = next_safe_pos = end;
 369
 370 out:
 371   // done.
 372   list<Context*> ls;
 373   ls.swap(waitfor_recover);
 374   finish_contexts(cct, ls, r);
 375 }
 376
 377 class Journaler::C_RereadHeadProbe : public Context
 378 {
 379   Journaler *ls;
 380   C_OnFinisher *final_finish;
 381 public:
 382   C_RereadHeadProbe(Journaler *l, C_OnFinisher *finish) :
 383     ls(l), final_finish(finish) {}
 384   void finish(int r) override {
 385     ls->_finish_reread_head_and_probe(r, final_finish);
 386   }
 387 };
 388
 389 void Journaler::reread_head_and_probe(Context *onfinish)
 390 {
 391   lock_guard l(lock);
 392
 393   assert(state == STATE_ACTIVE);
 394   _reread_head(new C_RereadHeadProbe(this, wrap_finisher(onfinish)));
 395 }
 396
 397 void Journaler::_finish_reread_head_and_probe(int r, C_OnFinisher *onfinish)
 398 {
 399   // Expect to be called back from finish_reread_head, which already takes lock
 400   // lock is locked
 401
 402   assert(!r); //if we get an error, we're boned
 403   _reprobe(onfinish);
 404 }
 405
 406
 407 // WRITING
 408
 409 class Journaler::C_WriteHead : public Context {
 410 public:
 411   Journaler *ls;
 412   Header h;
 413   C_OnFinisher *oncommit;
 414   C_WriteHead(Journaler *l, Header& h_, C_OnFinisher *c) : ls(l), h(h_),
 415                                                            oncommit(c) {}
 416   void finish(int r) override {
 417     ls->_finish_write_head(r, h, oncommit);
 418   }
 419 };
 420
 421 void Journaler::write_head(Context *oncommit)
 422 {
 423   lock_guard l(lock);
 424   _write_head(oncommit);
 425 }
 426
 427
 428 void Journaler::_write_head(Context *oncommit)
 429 {
 430   assert(!readonly);
 431   assert(state == STATE_ACTIVE);
 432   last_written.trimmed_pos = trimmed_pos;
 433   last_written.expire_pos = expire_pos;
 434   last_written.unused_field = expire_pos;
 435   last_written.write_pos = safe_pos;
 436   last_written.stream_format = stream_format;
 437   ldout(cct, 10) << "write_head " << last_written << dendl;
 438
 439   // Avoid persisting bad pointers in case of bugs
 440   assert(last_written.write_pos >= last_written.expire_pos);
 441   assert(last_written.expire_pos >= last_written.trimmed_pos);
 442
 443   last_wrote_head = ceph::real_clock::now();
 444
 445   bufferlist bl;
 446   ::encode(last_written, bl);
 447   SnapContext snapc;
 448
 449   object_t oid = file_object_t(ino, 0);
 450   object_locator_t oloc(pg_pool);
 451   objecter->write_full(oid, oloc, snapc, bl, ceph::real_clock::now(), 0,
 452                        wrap_finisher(new C_WriteHead(
 453                                              this, last_written,
 454                                              wrap_finisher(oncommit))),
 455                        0, 0, write_iohint);
 456 }
 457
 458 void Journaler::_finish_write_head(int r, Header &wrote,
 459                                    C_OnFinisher *oncommit)
 460 {
 461   lock_guard l(lock);
 462
 463   if (r < 0) {
 464     lderr(cct) << "_finish_write_head got " << cpp_strerror(r) << dendl;
 465     handle_write_error(r);
 466     return;
 467   }
 468   assert(!readonly);
 469   ldout(cct, 10) << "_finish_write_head " << wrote << dendl;
 470   last_committed = wrote;
 471   if (oncommit) {
 472     oncommit->complete(r);
 473   }
 474
 475   _trim();  // trim?
 476 }
 477
 478
 479 /***************** WRITING *******************/
 480
 481 class Journaler::C_Flush : public Context {
 482   Journaler *ls;
 483   uint64_t start;
 484   ceph::real_time stamp;
 485 public:
 486   C_Flush(Journaler *l, int64_t s, ceph::real_time st)
 487     : ls(l), start(s), stamp(st) {}
 488   void finish(int r) override {
 489     ls->_finish_flush(r, start, stamp);
 490   }
 491 };
 492
 493 void Journaler::_finish_flush(int r, uint64_t start, ceph::real_time stamp)
 494 {
 495   lock_guard l(lock);
 496   assert(!readonly);
 497
 498   if (r < 0) {
 499     lderr(cct) << "_finish_flush got " << cpp_strerror(r) << dendl;
 500     handle_write_error(r);
 501     return;
 502   }
 503
 504   assert(start < flush_pos);
 505
 506   // calc latency?
 507   if (logger) {
 508     ceph::timespan lat = ceph::real_clock::now() - stamp;
 509     logger->tinc(logger_key_lat, lat);
 510   }
 511
 512   // adjust safe_pos
 513   auto it = pending_safe.find(start);
 514   assert(it != pending_safe.end());
 515   pending_safe.erase(it);
 516   if (pending_safe.empty())
 517     safe_pos = next_safe_pos;
 518   else
 519     safe_pos = pending_safe.begin()->second;
 520
 521   ldout(cct, 10) << "_finish_flush safe from " << start
 522                  << ", pending_safe " << pending_safe
 523                  << ", (prezeroing/prezero)/write/flush/safe positions now "
 524                  << "(" << prezeroing_pos << "/" << prezero_pos << ")/"
 525                  << write_pos << "/" << flush_pos << "/" << safe_pos
 526                  << dendl;
 527
 528   // kick waiters <= safe_pos
 529   if (!waitfor_safe.empty()) {
 530     list<Context*> ls;
 531     while (!waitfor_safe.empty()) {
 532       auto it = waitfor_safe.begin();
 533       if (it->first > safe_pos)
 534         break;
 535       ls.splice(ls.end(), it->second);
 536       waitfor_safe.erase(it);
 537     }
 538     finish_contexts(cct, ls);
 539   }
 540 }
 541
 542
 543
 544 uint64_t Journaler::append_entry(bufferlist& bl)
 545 {
 546   unique_lock l(lock);
 547
 548   assert(!readonly);
 549   uint32_t s = bl.length();
 550
 551   // append
 552   size_t delta = bl.length() + journal_stream.get_envelope_size();
 553   // write_buf space is nearly full
 554   if (!write_buf_throttle.get_or_fail(delta)) {
 555     l.unlock();
 556     ldout(cct, 10) << "write_buf_throttle wait, delta " << delta << dendl;
 557     write_buf_throttle.get(delta);
 558     l.lock();
 559   }
 560   ldout(cct, 20) << "write_buf_throttle get, delta " << delta << dendl;
 561   size_t wrote = journal_stream.write(bl, &write_buf, write_pos);
 562   ldout(cct, 10) << "append_entry len " << s << " to " << write_pos << "~"
 563                  << wrote << dendl;
 564   write_pos += wrote;
 565
 566   // flush previous object?
 567   uint64_t su = get_layout_period();
 568   assert(su > 0);
 569   uint64_t write_off = write_pos % su;
 570   uint64_t write_obj = write_pos / su;
 571   uint64_t flush_obj = flush_pos / su;
 572   if (write_obj != flush_obj) {
 573     ldout(cct, 10) << " flushing completed object(s) (su " << su << " wro "
 574                    << write_obj << " flo " << flush_obj << ")" << dendl;
 575     _do_flush(write_buf.length() - write_off);
 576
 577     // if _do_flush() skips flushing some data, it does do a best effort to
 578     // update next_safe_pos.
 579     if (write_buf.length() > 0 &&
 580         write_buf.length() <= wrote) { // the unflushed data are within this entry
 581       // set next_safe_pos to end of previous entry
 582       next_safe_pos = write_pos - wrote;
 583     }
 584   }
 585
 586   return write_pos;
 587 }
 588
 589
 590 void Journaler::_do_flush(unsigned amount)
 591 {
 592   if (write_pos == flush_pos)
 593     return;
 594   assert(write_pos > flush_pos);
 595   assert(!readonly);
 596
 597   // flush
 598   uint64_t len = write_pos - flush_pos;
 599   assert(len == write_buf.length());
 600   if (amount && amount < len)
 601     len = amount;
 602
 603   // zero at least two full periods ahead.  this ensures
 604   // that the next object will not exist.
 605   uint64_t period = get_layout_period();
 606   if (flush_pos + len + 2*period > prezero_pos) {
 607     _issue_prezero();
 608
 609     int64_t newlen = prezero_pos - flush_pos - period;
 610     if (newlen <= 0) {
 611       ldout(cct, 10) << "_do_flush wanted to do " << flush_pos << "~" << len
 612                      << " already too close to prezero_pos " << prezero_pos
 613                      << ", zeroing first" << dendl;
 614       waiting_for_zero = true;
 615       return;
 616     }
 617     if (static_cast<uint64_t>(newlen) < len) {
 618       ldout(cct, 10) << "_do_flush wanted to do " << flush_pos << "~" << len
 619                      << " but hit prezero_pos " << prezero_pos
 620                      << ", will do " << flush_pos << "~" << newlen << dendl;
 621       len = newlen;
 622     } else {
 623       waiting_for_zero = false;
 624     }
 625   } else {
 626     waiting_for_zero = false;
 627   }
 628   ldout(cct, 10) << "_do_flush flushing " << flush_pos << "~" << len << dendl;
 629
 630   // submit write for anything pending
 631   // flush _start_ pos to _finish_flush
 632   ceph::real_time now = ceph::real_clock::now();
 633   SnapContext snapc;
 634
 635   Context *onsafe = new C_Flush(this, flush_pos, now);  // on COMMIT
 636   pending_safe[flush_pos] = next_safe_pos;
 637
 638   bufferlist write_bl;
 639
 640   // adjust pointers
 641   if (len == write_buf.length()) {
 642     write_bl.swap(write_buf);
 643     next_safe_pos = write_pos;
 644   } else {
 645     write_buf.splice(0, len, &write_bl);
 646     // Keys of waitfor_safe map are journal entry boundaries.
 647     // Try finding a journal entry that we are actually flushing
 648     // and set next_safe_pos to end of it. This is best effort.
 649     // The one we found may not be the lastest flushing entry.
 650     auto p = waitfor_safe.lower_bound(flush_pos + len);
 651     if (p != waitfor_safe.end()) {
 652       if (p->first > flush_pos + len && p != waitfor_safe.begin())
 653        --p;
 654       if (p->first <= flush_pos + len && p->first > next_safe_pos)
 655        next_safe_pos = p->first;
 656     }
 657   }
 658
 659   filer.write(ino, &layout, snapc,
 660               flush_pos, len, write_bl, ceph::real_clock::now(),
 661               0,
 662               wrap_finisher(onsafe), write_iohint);
 663
 664   flush_pos += len;
 665   assert(write_buf.length() == write_pos - flush_pos);
 666   write_buf_throttle.put(len);
 667   ldout(cct, 20) << "write_buf_throttle put, len " << len << dendl;
 668
 669   ldout(cct, 10)
 670     << "_do_flush (prezeroing/prezero)/write/flush/safe pointers now at "
 671     << "(" << prezeroing_pos << "/" << prezero_pos << ")/" << write_pos
 672     << "/" << flush_pos << "/" << safe_pos << dendl;
 673
 674   _issue_prezero();
 675 }
 676
 677
 678 void Journaler::wait_for_flush(Context *onsafe)
 679 {
 680   lock_guard l(lock);
 681   if (stopping) {
 682     onsafe->complete(-EAGAIN);
 683     return;
 684   }
 685   _wait_for_flush(onsafe);
 686 }
 687
 688 void Journaler::_wait_for_flush(Context *onsafe)
 689 {
 690   assert(!readonly);
 691
 692   // all flushed and safe?
 693   if (write_pos == safe_pos) {
 694     assert(write_buf.length() == 0);
 695     ldout(cct, 10)
 696       << "flush nothing to flush, (prezeroing/prezero)/write/flush/safe "
 697       "pointers at " << "(" << prezeroing_pos << "/" << prezero_pos << ")/"
 698       << write_pos << "/" << flush_pos << "/" << safe_pos << dendl;
 699     if (onsafe) {
 700       finisher->queue(onsafe, 0);
 701     }
 702     return;
 703   }
 704
 705   // queue waiter
 706   if (onsafe) {
 707     waitfor_safe[write_pos].push_back(wrap_finisher(onsafe));
 708   }
 709 }
 710
 711 void Journaler::flush(Context *onsafe)
 712 {
 713   lock_guard l(lock);
 714   _flush(wrap_finisher(onsafe));
 715 }
 716
 717 void Journaler::_flush(C_OnFinisher *onsafe)
 718 {
 719   assert(!readonly);
 720
 721   if (write_pos == flush_pos) {
 722     assert(write_buf.length() == 0);
 723     ldout(cct, 10) << "flush nothing to flush, (prezeroing/prezero)/write/"
 724       "flush/safe pointers at " << "(" << prezeroing_pos << "/" << prezero_pos
 725                    << ")/" << write_pos << "/" << flush_pos << "/" << safe_pos
 726                    << dendl;
 727     if (onsafe) {
 728       onsafe->complete(0);
 729     }
 730   } else {
 731     _do_flush();
 732     _wait_for_flush(onsafe);
 733   }
 734
 735   // write head?
 736   if (_write_head_needed()) {
 737     _write_head();
 738   }
 739 }
 740
 741 bool Journaler::_write_head_needed()
 742 {
 743   return last_wrote_head + seconds(cct->_conf->journaler_write_head_interval)
 744       < ceph::real_clock::now();
 745 }
 746
 747
 748 /*************** prezeroing ******************/
 749
 750 struct C_Journaler_Prezero : public Context {
 751   Journaler *journaler;
 752   uint64_t from, len;
 753   C_Journaler_Prezero(Journaler *j, uint64_t f, uint64_t l)
 754     : journaler(j), from(f), len(l) {}
 755   void finish(int r) override {
 756     journaler->_finish_prezero(r, from, len);
 757   }
 758 };
 759
 760 void Journaler::_issue_prezero()
 761 {
 762   assert(prezeroing_pos >= flush_pos);
 763
 764   // we need to zero at least two periods, minimum, to ensure that we
 765   // have a full empty object/period in front of us.
 766   uint64_t num_periods = MAX(2, cct->_conf->journaler_prezero_periods);
 767
 768   /*
 769    * issue zero requests based on write_pos, even though the invariant
 770    * is that we zero ahead of flush_pos.
 771    */
 772   uint64_t period = get_layout_period();
 773   uint64_t to = write_pos + period * num_periods  + period - 1;
 774   to -= to % period;
 775
 776   if (prezeroing_pos >= to) {
 777     ldout(cct, 20) << "_issue_prezero target " << to << " <= prezeroing_pos "
 778                    << prezeroing_pos << dendl;
 779     return;
 780   }
 781
 782   while (prezeroing_pos < to) {
 783     uint64_t len;
 784     if (prezeroing_pos % period == 0) {
 785       len = period;
 786       ldout(cct, 10) << "_issue_prezero removing " << prezeroing_pos << "~"
 787                      << period << " (full period)" << dendl;
 788     } else {
 789       len = period - (prezeroing_pos % period);
 790       ldout(cct, 10) << "_issue_prezero zeroing " << prezeroing_pos << "~"
 791                      << len << " (partial period)" << dendl;
 792     }
 793     SnapContext snapc;
 794     Context *c = wrap_finisher(new C_Journaler_Prezero(this, prezeroing_pos,
 795                                                        len));
 796     filer.zero(ino, &layout, snapc, prezeroing_pos, len,
 797                ceph::real_clock::now(), 0, c);
 798     prezeroing_pos += len;
 799   }
 800 }
 801
 802 // Lock cycle because we get called out of objecter callback (holding
 803 // objecter read lock), but there are also cases where we take the journaler
 804 // lock before calling into objecter to do I/O.
 805 void Journaler::_finish_prezero(int r, uint64_t start, uint64_t len)
 806 {
 807   lock_guard l(lock);
 808
 809   ldout(cct, 10) << "_prezeroed to " << start << "~" << len
 810                  << ", prezeroing/prezero was " << prezeroing_pos << "/"
 811                  << prezero_pos << ", pending " << pending_zero
 812                  << dendl;
 813   if (r < 0 && r != -ENOENT) {
 814     lderr(cct) << "_prezeroed got " << cpp_strerror(r) << dendl;
 815     handle_write_error(r);
 816     return;
 817   }
 818
 819   assert(r == 0 || r == -ENOENT);
 820
 821   if (start == prezero_pos) {
 822     prezero_pos += len;
 823     while (!pending_zero.empty() &&
 824            pending_zero.begin().get_start() == prezero_pos) {
 825       interval_set<uint64_t>::iterator b(pending_zero.begin());
 826       prezero_pos += b.get_len();
 827       pending_zero.erase(b);
 828     }
 829
 830     if (waiting_for_zero) {
 831       _do_flush();
 832     }
 833   } else {
 834     pending_zero.insert(start, len);
 835   }
 836   ldout(cct, 10) << "_prezeroed prezeroing/prezero now " << prezeroing_pos
 837                  << "/" << prezero_pos
 838                  << ", pending " << pending_zero
 839                  << dendl;
 840 }
 841
 842
 843
 844 /***************** READING *******************/
 845
 846
 847 class Journaler::C_Read : public Context {
 848   Journaler *ls;
 849   uint64_t offset;
 850   uint64_t length;
 851 public:
 852   bufferlist bl;
 853   C_Read(Journaler *j, uint64_t o, uint64_t l) : ls(j), offset(o), length(l) {}
 854   void finish(int r) override {
 855     ls->_finish_read(r, offset, length, bl);
 856   }
 857 };
 858
 859 class Journaler::C_RetryRead : public Context {
 860   Journaler *ls;
 861 public:
 862   explicit C_RetryRead(Journaler *l) : ls(l) {}
 863
 864   void finish(int r) override {
 865     // Should only be called from waitfor_safe i.e. already inside lock
 866     // (ls->lock is locked
 867     ls->_prefetch();
 868   }
 869 };
 870
 871 void Journaler::_finish_read(int r, uint64_t offset, uint64_t length,
 872                              bufferlist& bl)
 873 {
 874   lock_guard l(lock);
 875
 876   if (r < 0) {
 877     ldout(cct, 0) << "_finish_read got error " << r << dendl;
 878     error = r;
 879   } else {
 880     ldout(cct, 10) << "_finish_read got " << offset << "~" << bl.length()
 881                    << dendl;
 882     if (bl.length() < length) {
 883       ldout(cct, 0) << "_finish_read got less than expected (" << length << ")"
 884                     << dendl;
 885       error = -EINVAL;
 886     }
 887   }
 888
 889   if (error) {
 890     if (on_readable) {
 891       C_OnFinisher *f = on_readable;
 892       on_readable = 0;
 893       f->complete(error);
 894     }
 895     return;
 896   }
 897
 898   prefetch_buf[offset].swap(bl);
 899
 900   try {
 901     _assimilate_prefetch();
 902   } catch (const buffer::error &err) {
 903     lderr(cct) << "_decode error from assimilate_prefetch" << dendl;
 904     error = -EINVAL;
 905     if (on_readable) {
 906       C_OnFinisher *f = on_readable;
 907       on_readable = 0;
 908       f->complete(error);
 909     }
 910     return;
 911   }
 912   _prefetch();
 913 }
 914
 915 void Journaler::_assimilate_prefetch()
 916 {
 917   bool was_readable = readable;
 918
 919   bool got_any = false;
 920   while (!prefetch_buf.empty()) {
 921     map<uint64_t,bufferlist>::iterator p = prefetch_buf.begin();
 922     if (p->first != received_pos) {
 923       uint64_t gap = p->first - received_pos;
 924       ldout(cct, 10) << "_assimilate_prefetch gap of " << gap
 925                      << " from received_pos " << received_pos
 926                      << " to first prefetched buffer " << p->first << dendl;
 927       break;
 928     }
 929
 930     ldout(cct, 10) << "_assimilate_prefetch " << p->first << "~"
 931                    << p->second.length() << dendl;
 932     received_pos += p->second.length();
 933     read_buf.claim_append(p->second);
 934     assert(received_pos <= requested_pos);
 935     prefetch_buf.erase(p);
 936     got_any = true;
 937   }
 938
 939   if (got_any) {
 940     ldout(cct, 10) << "_assimilate_prefetch read_buf now " << read_pos << "~"
 941                    << read_buf.length() << ", read pointers " << read_pos
 942                    << "/" << received_pos << "/" << requested_pos
 943                    << dendl;
 944
 945     // Update readability (this will also hit any decode errors resulting
 946     // from bad data)
 947     readable = _is_readable();
 948   }
 949
 950   if ((got_any && !was_readable && readable) || read_pos == write_pos) {
 951     // readable!
 952     ldout(cct, 10) << "_finish_read now readable (or at journal end) readable="
 953                    << readable << " read_pos=" << read_pos << " write_pos="
 954                    << write_pos << dendl;
 955     if (on_readable) {
 956       C_OnFinisher *f = on_readable;
 957       on_readable = 0;
 958       f->complete(0);
 959     }
 960   }
 961 }
 962
 963 void Journaler::_issue_read(uint64_t len)
 964 {
 965   // stuck at safe_pos?  (this is needed if we are reading the tail of
 966   // a journal we are also writing to)
 967   assert(requested_pos <= safe_pos);
 968   if (requested_pos == safe_pos) {
 969     ldout(cct, 10) << "_issue_read requested_pos = safe_pos = " << safe_pos
 970                    << ", waiting" << dendl;
 971     assert(write_pos > requested_pos);
 972     if (pending_safe.empty()) {
 973       _flush(NULL);
 974     }
 975
 976     // Make sure keys of waitfor_safe map are journal entry boundaries.
 977     // The key we used here is either next_safe_pos or old value of
 978     // next_safe_pos. next_safe_pos is always set to journal entry
 979     // boundary.
 980     auto p = pending_safe.rbegin();
 981     if (p != pending_safe.rend())
 982       waitfor_safe[p->second].push_back(new C_RetryRead(this));
 983     else
 984       waitfor_safe[next_safe_pos].push_back(new C_RetryRead(this));
 985     return;
 986   }
 987
 988   // don't read too much
 989   if (requested_pos + len > safe_pos) {
 990     len = safe_pos - requested_pos;
 991     ldout(cct, 10) << "_issue_read reading only up to safe_pos " << safe_pos
 992                    << dendl;
 993   }
 994
 995   // go.
 996   ldout(cct, 10) << "_issue_read reading " << requested_pos << "~" << len
 997                  << ", read pointers " << read_pos << "/" << received_pos
 998                  << "/" << (requested_pos+len) << dendl;
 999
1000   // step by period (object).  _don't_ do a single big filer.read()
1001   // here because it will wait for all object reads to complete before
1002   // giving us back any data.  this way we can process whatever bits
1003   // come in that are contiguous.
1004   uint64_t period = get_layout_period();
1005   while (len > 0) {
1006     uint64_t e = requested_pos + period;
1007     e -= e % period;
1008     uint64_t l = e - requested_pos;
1009     if (l > len)
1010       l = len;
1011     C_Read *c = new C_Read(this, requested_pos, l);
1012     filer.read(ino, &layout, CEPH_NOSNAP, requested_pos, l, &c->bl, 0,
1013                wrap_finisher(c), CEPH_OSD_OP_FLAG_FADVISE_DONTNEED);
1014     requested_pos += l;
1015     len -= l;
1016   }
1017 }
1018
1019 void Journaler::_prefetch()
1020 {
1021   ldout(cct, 10) << "_prefetch" << dendl;
1022   // prefetch
1023   uint64_t pf;
1024   if (temp_fetch_len) {
1025     ldout(cct, 10) << "_prefetch temp_fetch_len " << temp_fetch_len << dendl;
1026     pf = temp_fetch_len;
1027     temp_fetch_len = 0;
1028   } else {
1029     pf = fetch_len;
1030   }
1031
1032   uint64_t raw_target = read_pos + pf;
1033
1034   // read full log segments, so increase if necessary
1035   uint64_t period = get_layout_period();
1036   uint64_t remainder = raw_target % period;
1037   uint64_t adjustment = remainder ? period - remainder : 0;
1038   uint64_t target = raw_target + adjustment;
1039
1040   // don't read past the log tail
1041   if (target > write_pos)
1042     target = write_pos;
1043
1044   if (requested_pos < target) {
1045     uint64_t len = target - requested_pos;
1046     ldout(cct, 10) << "_prefetch " << pf << " requested_pos " << requested_pos
1047                    << " < target " << target << " (" << raw_target
1048                    << "), prefetching " << len << dendl;
1049
1050     if (pending_safe.empty() && write_pos > safe_pos) {
1051       // If we are reading and writing the journal, then we may need
1052       // to issue a flush if one isn't already in progress.
1053       // Avoid doing a flush every time so that if we do write/read/write/read
1054       // we don't end up flushing after every write.
1055       ldout(cct, 10) << "_prefetch: requested_pos=" << requested_pos
1056                      << ", read_pos=" << read_pos
1057                      << ", write_pos=" << write_pos
1058                      << ", safe_pos=" << safe_pos << dendl;
1059       _do_flush();
1060     }
1061
1062     _issue_read(len);
1063   }
1064 }
1065
1066
1067 /*
1068  * _is_readable() - return true if next entry is ready.
1069  */
1070 bool Journaler::_is_readable()
1071 {
1072   // anything to read?
1073   if (read_pos == write_pos)
1074     return false;
1075
1076   // Check if the retrieve bytestream has enough for an entry
1077   uint64_t need;
1078   if (journal_stream.readable(read_buf, &need)) {
1079     return true;
1080   }
1081
1082   ldout (cct, 10) << "_is_readable read_buf.length() == " << read_buf.length()
1083                   << ", but need " << need << " for next entry; fetch_len is "
1084                   << fetch_len << dendl;
1085
1086   // partial fragment at the end?
1087   if (received_pos == write_pos) {
1088     ldout(cct, 10) << "is_readable() detected partial entry at tail, "
1089       "adjusting write_pos to " << read_pos << dendl;
1090
1091     // adjust write_pos
1092     prezeroing_pos = prezero_pos = write_pos = flush_pos = safe_pos = next_safe_pos = read_pos;
1093     assert(write_buf.length() == 0);
1094     assert(waitfor_safe.empty());
1095
1096     // reset read state
1097     requested_pos = received_pos = read_pos;
1098     read_buf.clear();
1099
1100     // FIXME: truncate on disk?
1101
1102     return false;
1103   }
1104
1105   if (need > fetch_len) {
1106     temp_fetch_len = need;
1107     ldout(cct, 10) << "_is_readable noting temp_fetch_len " << temp_fetch_len
1108                    << dendl;
1109   }
1110
1111   ldout(cct, 10) << "_is_readable: not readable, returning false" << dendl;
1112   return false;
1113 }
1114
1115 /*
1116  * is_readable() - kickstart prefetch, too
1117  */
1118 bool Journaler::is_readable()
1119 {
1120   lock_guard l(lock);
1121
1122   if (error != 0) {
1123     return false;
1124   }
1125
1126   bool r = readable;
1127   _prefetch();
1128   return r;
1129 }
1130
1131 class Journaler::C_EraseFinish : public Context {
1132   Journaler *journaler;
1133   C_OnFinisher *completion;
1134   public:
1135   C_EraseFinish(Journaler *j, C_OnFinisher *c) : journaler(j), completion(c) {}
1136   void finish(int r) override {
1137     journaler->_finish_erase(r, completion);
1138   }
1139 };
1140
1141 /**
1142  * Entirely erase the journal, including header.  For use when you
1143  * have already made a copy of the journal somewhere else.
1144  */
1145 void Journaler::erase(Context *completion)
1146 {
1147   lock_guard l(lock);
1148
1149   // Async delete the journal data
1150   uint64_t first = trimmed_pos / get_layout_period();
1151   uint64_t num = (write_pos - trimmed_pos) / get_layout_period() + 2;
1152   filer.purge_range(ino, &layout, SnapContext(), first, num,
1153                     ceph::real_clock::now(), 0,
1154                     wrap_finisher(new C_EraseFinish(
1155                                     this, wrap_finisher(completion))));
1156
1157   // We will not start the operation to delete the header until
1158   // _finish_erase has seen the data deletion succeed: otherwise if
1159   // there was an error deleting data we might prematurely delete the
1160   // header thereby lose our reference to the data.
1161 }
1162
1163 void Journaler::_finish_erase(int data_result, C_OnFinisher *completion)
1164 {
1165   lock_guard l(lock);
1166
1167   if (data_result == 0) {
1168     // Async delete the journal header
1169     filer.purge_range(ino, &layout, SnapContext(), 0, 1,
1170                       ceph::real_clock::now(),
1171                       0, wrap_finisher(completion));
1172   } else {
1173     lderr(cct) << "Failed to delete journal " << ino << " data: "
1174                << cpp_strerror(data_result) << dendl;
1175     completion->complete(data_result);
1176   }
1177 }
1178
1179 /* try_read_entry(bl)
1180  *  read entry into bl if it's ready.
1181  *  otherwise, do nothing.
1182  */
1183 bool Journaler::try_read_entry(bufferlist& bl)
1184 {
1185   lock_guard l(lock);
1186
1187   if (!readable) {
1188     ldout(cct, 10) << "try_read_entry at " << read_pos << " not readable"
1189                    << dendl;
1190     return false;
1191   }
1192
1193   uint64_t start_ptr;
1194   size_t consumed;
1195   try {
1196     consumed = journal_stream.read(read_buf, &bl, &start_ptr);
1197     if (stream_format >= JOURNAL_FORMAT_RESILIENT) {
1198       assert(start_ptr == read_pos);
1199     }
1200   } catch (const buffer::error &e) {
1201     lderr(cct) << __func__ << ": decode error from journal_stream" << dendl;
1202     error = -EINVAL;
1203     return false;
1204   }
1205
1206   ldout(cct, 10) << "try_read_entry at " << read_pos << " read "
1207                  << read_pos << "~" << consumed << " (have "
1208                  << read_buf.length() << ")" << dendl;
1209
1210   read_pos += consumed;
1211   try {
1212     // We were readable, we might not be any more
1213     readable = _is_readable();
1214   } catch (const buffer::error &e) {
1215     lderr(cct) << __func__ << ": decode error from _is_readable" << dendl;
1216     error = -EINVAL;
1217     return false;
1218   }
1219
1220   // prefetch?
1221   _prefetch();
1222   return true;
1223 }
1224
1225 void Journaler::wait_for_readable(Context *onreadable)
1226 {
1227   lock_guard l(lock);
1228   if (stopping) {
1229     finisher->queue(onreadable, -EAGAIN);
1230     return;
1231   }
1232
1233   assert(on_readable == 0);
1234   if (!readable) {
1235     ldout(cct, 10) << "wait_for_readable at " << read_pos << " onreadable "
1236                    << onreadable << dendl;
1237     on_readable = wrap_finisher(onreadable);
1238   } else {
1239     // race with OSD reply
1240     finisher->queue(onreadable, 0);
1241   }
1242 }
1243
1244 bool Journaler::have_waiter() const
1245 {
1246   return on_readable != nullptr;
1247 }
1248
1249
1250
1251
1252 /***************** TRIMMING *******************/
1253
1254
1255 class Journaler::C_Trim : public Context {
1256   Journaler *ls;
1257   uint64_t to;
1258 public:
1259   C_Trim(Journaler *l, int64_t t) : ls(l), to(t) {}
1260   void finish(int r) override {
1261     ls->_finish_trim(r, to);
1262   }
1263 };
1264
1265 void Journaler::trim()
1266 {
1267   lock_guard l(lock);
1268   _trim();
1269 }
1270
1271 void Journaler::_trim()
1272 {
1273   assert(!readonly);
1274   uint64_t period = get_layout_period();
1275   uint64_t trim_to = last_committed.expire_pos;
1276   trim_to -= trim_to % period;
1277   ldout(cct, 10) << "trim last_commited head was " << last_committed
1278            << ", can trim to " << trim_to
1279            << dendl;
1280   if (trim_to == 0 || trim_to == trimming_pos) {
1281     ldout(cct, 10) << "trim already trimmed/trimming to "
1282                    << trimmed_pos << "/" << trimming_pos << dendl;
1283     return;
1284   }
1285
1286   if (trimming_pos > trimmed_pos) {
1287     ldout(cct, 10) << "trim already trimming atm, try again later.  "
1288       "trimmed/trimming is " << trimmed_pos << "/" << trimming_pos << dendl;
1289     return;
1290   }
1291
1292   // trim
1293   assert(trim_to <= write_pos);
1294   assert(trim_to <= expire_pos);
1295   assert(trim_to > trimming_pos);
1296   ldout(cct, 10) << "trim trimming to " << trim_to
1297                  << ", trimmed/trimming/expire are "
1298                  << trimmed_pos << "/" << trimming_pos << "/" << expire_pos
1299                  << dendl;
1300
1301   // delete range of objects
1302   uint64_t first = trimming_pos / period;
1303   uint64_t num = (trim_to - trimming_pos) / period;
1304   SnapContext snapc;
1305   filer.purge_range(ino, &layout, snapc, first, num,
1306                     ceph::real_clock::now(), 0,
1307                     wrap_finisher(new C_Trim(this, trim_to)));
1308   trimming_pos = trim_to;
1309 }
1310
1311 void Journaler::_finish_trim(int r, uint64_t to)
1312 {
1313   lock_guard l(lock);
1314
1315   assert(!readonly);
1316   ldout(cct, 10) << "_finish_trim trimmed_pos was " << trimmed_pos
1317            << ", trimmed/trimming/expire now "
1318            << to << "/" << trimming_pos << "/" << expire_pos
1319            << dendl;
1320   if (r < 0 && r != -ENOENT) {
1321     lderr(cct) << "_finish_trim got " << cpp_strerror(r) << dendl;
1322     handle_write_error(r);
1323     return;
1324   }
1325
1326   assert(r >= 0 || r == -ENOENT);
1327
1328   assert(to <= trimming_pos);
1329   assert(to > trimmed_pos);
1330   trimmed_pos = to;
1331 }
1332
1333 void Journaler::handle_write_error(int r)
1334 {
1335   // lock is locked
1336
1337   lderr(cct) << "handle_write_error " << cpp_strerror(r) << dendl;
1338   if (on_write_error) {
1339     on_write_error->complete(r);
1340     on_write_error = NULL;
1341     called_write_error = true;
1342   } else if (called_write_error) {
1343     /* We don't call error handler more than once, subsequent errors
1344      * are dropped -- this is okay as long as the error handler does
1345      * something dramatic like respawn */
1346     lderr(cct) << __func__ << ": multiple write errors, handler already called"
1347                << dendl;
1348   } else {
1349     assert(0 == "unhandled write error");
1350   }
1351 }
1352
1353
1354 /**
1355  * Test whether the 'read_buf' byte stream has enough data to read
1356  * an entry
1357  *
1358  * sets 'next_envelope_size' to the number of bytes needed to advance (enough
1359  * to get the next header if header was unavailable, or enough to get the whole
1360  * next entry if the header was available but the body wasn't).
1361  */
1362 bool JournalStream::readable(bufferlist &read_buf, uint64_t *need) const
1363 {
1364   assert(need != NULL);
1365
1366   uint32_t entry_size = 0;
1367   uint64_t entry_sentinel = 0;
1368   bufferlist::iterator p = read_buf.begin();
1369
1370   // Do we have enough data to decode an entry prefix?
1371   if (format >= JOURNAL_FORMAT_RESILIENT) {
1372     *need = sizeof(entry_size) + sizeof(entry_sentinel);
1373   } else {
1374     *need = sizeof(entry_size);
1375   }
1376   if (read_buf.length() >= *need) {
1377     if (format >= JOURNAL_FORMAT_RESILIENT) {
1378       ::decode(entry_sentinel, p);
1379       if (entry_sentinel != sentinel) {
1380         throw buffer::malformed_input("Invalid sentinel");
1381       }
1382     }
1383
1384     ::decode(entry_size, p);
1385   } else {
1386     return false;
1387   }
1388
1389   // Do we have enough data to decode an entry prefix, payload and suffix?
1390   if (format >= JOURNAL_FORMAT_RESILIENT) {
1391     *need = JOURNAL_ENVELOPE_RESILIENT + entry_size;
1392   } else {
1393     *need = JOURNAL_ENVELOPE_LEGACY + entry_size;
1394   }
1395   if (read_buf.length() >= *need) {
1396     return true;  // No more bytes needed
1397   }
1398
1399   return false;
1400 }
1401
1402
1403 /**
1404  * Consume one entry from a journal byte stream 'from', splicing a
1405  * serialized LogEvent blob into 'entry'.
1406  *
1407  * 'entry' must be non null and point to an empty bufferlist.
1408  *
1409  * 'from' must contain sufficient valid data (i.e. readable is true).
1410  *
1411  * 'start_ptr' will be set to the entry's start pointer, if the collection
1412  * format provides it.  It may not be null.
1413  *
1414  * @returns The number of bytes consumed from the `from` byte stream.  Note
1415  *          that this is not equal to the length of `entry`, which contains
1416  *          the inner serialized LogEvent and not the envelope.
1417  */
1418 size_t JournalStream::read(bufferlist &from, bufferlist *entry,
1419                            uint64_t *start_ptr)
1420 {
1421   assert(start_ptr != NULL);
1422   assert(entry != NULL);
1423   assert(entry->length() == 0);
1424
1425   uint32_t entry_size = 0;
1426
1427   // Consume envelope prefix: entry_size and entry_sentinel
1428   bufferlist::iterator from_ptr = from.begin();
1429   if (format >= JOURNAL_FORMAT_RESILIENT) {
1430     uint64_t entry_sentinel = 0;
1431     ::decode(entry_sentinel, from_ptr);
1432     // Assertion instead of clean check because of precondition of this
1433     // fn is that readable() already passed
1434     assert(entry_sentinel == sentinel);
1435   }
1436   ::decode(entry_size, from_ptr);
1437
1438   // Read out the payload
1439   from_ptr.copy(entry_size, *entry);
1440
1441   // Consume the envelope suffix (start_ptr)
1442   if (format >= JOURNAL_FORMAT_RESILIENT) {
1443     ::decode(*start_ptr, from_ptr);
1444   } else {
1445     *start_ptr = 0;
1446   }
1447
1448   // Trim the input buffer to discard the bytes we have consumed
1449   from.splice(0, from_ptr.get_off());
1450
1451   return from_ptr.get_off();
1452 }
1453
1454
1455 /**
1456  * Append one entry
1457  */
1458 size_t JournalStream::write(bufferlist &entry, bufferlist *to,
1459                             uint64_t const &start_ptr)
1460 {
1461   assert(to != NULL);
1462
1463   uint32_t const entry_size = entry.length();
1464   if (format >= JOURNAL_FORMAT_RESILIENT) {
1465     ::encode(sentinel, *to);
1466   }
1467   ::encode(entry_size, *to);
1468   to->claim_append(entry);
1469   if (format >= JOURNAL_FORMAT_RESILIENT) {
1470     ::encode(start_ptr, *to);
1471   }
1472
1473   if (format >= JOURNAL_FORMAT_RESILIENT) {
1474     return JOURNAL_ENVELOPE_RESILIENT + entry_size;
1475   } else {
1476     return JOURNAL_ENVELOPE_LEGACY + entry_size;
1477   }
1478 }
1479
1480 /**
1481  * set write error callback
1482  *
1483  * Set a callback/context to trigger if we get a write error from
1484  * the objecter.  This may be from an explicit request (e.g., flush)
1485  * or something async the journaler did on its own (e.g., journal
1486  * header update).
1487  *
1488  * It is only used once; if the caller continues to use the
1489  * Journaler and wants to hear about errors, it needs to reset the
1490  * error_handler.
1491  *
1492  * @param c callback/context to trigger on error
1493  */
1494 void Journaler::set_write_error_handler(Context *c) {
1495   lock_guard l(lock);
1496   assert(!on_write_error);
1497   on_write_error = wrap_finisher(c);
1498   called_write_error = false;
1499 }
1500
1501
1502 /**
1503  * Wrap a context in a C_OnFinisher, if it is non-NULL
1504  *
1505  * Utility function to avoid lots of error-prone and verbose
1506  * NULL checking on contexts passed in.
1507  */
1508 C_OnFinisher *Journaler::wrap_finisher(Context *c)
1509 {
1510   if (c != NULL) {
1511     return new C_OnFinisher(c, finisher);
1512   } else {
1513     return NULL;
1514   }
1515 }
1516
1517 void Journaler::shutdown()
1518 {
1519   lock_guard l(lock);
1520
1521   ldout(cct, 1) << __func__ << dendl;
1522
1523   readable = false;
1524   stopping = true;
1525
1526   // Kick out anyone reading from journal
1527   error = -EAGAIN;
1528   if (on_readable) {
1529     C_OnFinisher *f = on_readable;
1530     on_readable = 0;
1531     f->complete(-EAGAIN);
1532   }
1533
1534   finish_contexts(cct, waitfor_recover, -ESHUTDOWN);
1535
1536   std::map<uint64_t, std::list<Context*> >::iterator i;
1537   for (i = waitfor_safe.begin(); i != waitfor_safe.end(); ++i) {
1538     finish_contexts(cct, i->second, -EAGAIN);
1539   }
1540   waitfor_safe.clear();
1541 }
1542