text: ////////////////////////////////////////////////////////////////////////////////
  /// DISCLAIMER
  ///
  /// Copyright 2014-2016 ArangoDB GmbH, Cologne, Germany
  /// Copyright 2004-2014 triAGENS GmbH, Cologne, Germany
  ///
  /// Licensed under the Apache License, Version 2.0 (the "License");
  /// you may not use this file except in compliance with the License.
  /// You may obtain a copy of the License at
  ///
  ///     http://www.apache.org/licenses/LICENSE-2.0
  ///
  /// Unless required by applicable law or agreed to in writing, software
  /// distributed under the License is distributed on an "AS IS" BASIS,
  /// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  /// See the License for the specific language governing permissions and
  /// limitations under the License.
  ///
  /// Copyright holder is ArangoDB GmbH, Cologne, Germany
  ///
  /// @author Jan Steemann
  ////////////////////////////////////////////////////////////////////////////////
  
  #include "MMFilesWalSlots.h"
  #include "ApplicationFeatures/ApplicationServer.h"
  #include "Basics/ConditionLocker.h"
  #include "Basics/MutexLocker.h"
  #include "Basics/encoding.h"
  #include "Basics/system-functions.h"
  #include "Logger/Logger.h"
  #include "MMFiles/MMFilesDatafile.h"
  #include "MMFiles/MMFilesDatafileHelper.h"
  #include "MMFiles/MMFilesLogfileManager.h"
  #include "VocBase/ticks.h"
  
  using namespace arangodb;
  
  static uint32_t const PrologueSize =
      encoding::alignedSize<uint32_t>(sizeof(MMFilesPrologueMarker));
  
  /// @brief create the slots
  MMFilesWalSlots::MMFilesWalSlots(MMFilesLogfileManager* logfileManager,
                                   size_t numberOfSlots, MMFilesWalSlot::TickType tick)
      : _logfileManager(logfileManager),
        _condition(),
        _lock(),
        _slots(nullptr),
        _numberOfSlots(numberOfSlots),
        _freeSlots(numberOfSlots),
        _waiting(0),
        _handoutIndex(0),
        _recycleIndex(0),
        _logfile(nullptr),
        _lastAssignedTick(0),
        _lastCommittedTick(0),
        _lastCommittedDataTick(0),
        _numEvents(0),
        _numEventsSync(0),
        _lastDatabaseId(0),
        _lastCollectionId(0),
        _shutdown(false) {
    _slots = new MMFilesWalSlot[numberOfSlots];
  }
  
  /// @brief destroy the slots
  MMFilesWalSlots::~MMFilesWalSlots() { delete[] _slots; }
  
  /// @brief sets a shutdown flag, disabling the request for new logfiles
  void MMFilesWalSlots::shutdown() {
    MUTEX_LOCKER(mutexLocker, _lock);
    _shutdown = true;
  }
  
  /// @brief get the statistics of the slots
  void MMFilesWalSlots::statistics(MMFilesWalSlot::TickType& lastAssignedTick,
                                   MMFilesWalSlot::TickType& lastCommittedTick,
                                   MMFilesWalSlot::TickType& lastCommittedDataTick,
                                   uint64_t& numEvents, uint64_t& numEventsSync) {
    MUTEX_LOCKER(mutexLocker, _lock);
    lastAssignedTick = _lastAssignedTick;
    lastCommittedTick = _lastCommittedTick;
    lastCommittedDataTick = _lastCommittedDataTick;
    numEvents = _numEvents;
    numEventsSync = _numEventsSync;
  }
  
  /// @brief initially set the last ticks on start
  void MMFilesWalSlots::setLastTick(MMFilesWalSlot::TickType const& tick) {
    MUTEX_LOCKER(mutexLocker, _lock);
    if (tick > _lastAssignedTick) {
      _lastAssignedTick = tick;
    }
    if (tick > _lastCommittedTick) {
      _lastCommittedTick = tick;
    }
    if (tick > _lastCommittedDataTick) {
      _lastCommittedDataTick = tick;
    }
  }
  
  /// @brief execute a flush operation
  int MMFilesWalSlots::flush(bool waitForSync) {
    MMFilesWalSlot::TickType lastTick = 0;
    bool worked;
  
    int res = closeLogfile(lastTick, worked);
  
    if (res == TRI_ERROR_REQUEST_CANCELED) {
      // only happens during shutdown
      res = TRI_ERROR_NO_ERROR;
    }
  
    if (res == TRI_ERROR_NO_ERROR) {
      _logfileManager->signalSync(true);
  
      if (waitForSync) {
        // wait until data has been committed to disk
        if (!waitForTick(lastTick)) {
          res = TRI_ERROR_ARANGO_SYNC_TIMEOUT;
        } else if (!worked) {
          res = TRI_ERROR_ARANGO_DATAFILE_EMPTY;
        }
      } else if (!worked) {
        // logfile to flush was still empty and thus not flushed
  
        // not really an error, but used to indicate the specific condition
        res = TRI_ERROR_ARANGO_DATAFILE_EMPTY;
      }
    }
  
    return res;
  }
  
  /// @brief return the last committed tick
  MMFilesWalSlot::TickType MMFilesWalSlots::lastCommittedTick() {
    MUTEX_LOCKER(mutexLocker, _lock);
    return _lastCommittedTick;
  }
  
  /// @brief return the next unused slot
  MMFilesWalSlotInfo MMFilesWalSlots::nextUnused(uint32_t size) {
    return nextUnused(0, 0, size);
  }
  
  /// @brief return the next unused slot
  MMFilesWalSlotInfo MMFilesWalSlots::nextUnused(TRI_voc_tick_t databaseId,
                                                 TRI_voc_cid_t collectionId, uint32_t size) {
    // we need to use the aligned size for writing
    uint32_t alignedSize = encoding::alignedSize<uint32_t>(size);
    int iterations = 0;
    bool hasWaited = false;
    bool mustWritePrologue = false;
  
    TRI_ASSERT(size > 0);
  
    while (++iterations < 1000) {
      {
        MUTEX_LOCKER(mutexLocker, _lock);
  
        MMFilesWalSlot* slot = &_slots[_handoutIndex];
        TRI_ASSERT(slot != nullptr);
  
        // check if next slot is free for writing
        // and also is the slot following it is also free for writing
        // this is required because in some cases we need two free slots
        // to write a WAL entry: the first slot for the prologue marker
        // and the second slot for the actual marker
        if (slot->isUnused() && _slots[nextHandoutIndex()].isUnused()) {
          if (hasWaited) {
            CONDITION_LOCKER(guard, _condition);
            TRI_ASSERT(_waiting > 0);
            --_waiting;
          }
  
          if (databaseId == 0 && collectionId == 0) {
            _lastDatabaseId = 0;
            _lastCollectionId = 0;
          } else if (!mustWritePrologue && databaseId > 0 && collectionId > 0 &&
                     (_lastDatabaseId != databaseId || _lastCollectionId != collectionId)) {
            // write a prologue
            alignedSize = size + PrologueSize;
            mustWritePrologue = true;
          }
  
          // cycle until we have a valid logfile
          while (_logfile == nullptr ||
                 _logfile->freeSize() < static_cast<uint64_t>(alignedSize)) {
            if (_logfile != nullptr) {
              // seal existing logfile by creating a footer marker
              int res = writeFooter(slot);
  
              if (res != TRI_ERROR_NO_ERROR) {
                return MMFilesWalSlotInfo(res);
              }
  
              // new datafile. must write a prologue
              if (databaseId > 0 && collectionId > 0 && !mustWritePrologue) {
                alignedSize = size + PrologueSize;
                mustWritePrologue = true;
              }
  
              // advance to next slot
              slot = &_slots[_handoutIndex];
              _logfileManager->setLogfileSealRequested(_logfile);
  
              _logfile = nullptr;
            }
  
            TRI_IF_FAILURE("LogfileManagerGetWriteableLogfile") {
              return MMFilesWalSlotInfo(TRI_ERROR_ARANGO_NO_JOURNAL);
            }
  
            // fetch the next free logfile (this may create a new one)
            MMFilesWalLogfile::StatusType status;
            int res = newLogfile(alignedSize, status);
  
            if (res != TRI_ERROR_NO_ERROR) {
              if (res != TRI_ERROR_ARANGO_NO_JOURNAL) {
                return MMFilesWalSlotInfo(res);
              }
  
              std::this_thread::sleep_for(std::chrono::milliseconds(10));
              // try again in next iteration
            } else {
              TRI_ASSERT(_logfile != nullptr);
  
              if (status == MMFilesWalLogfile::StatusType::EMPTY) {
                // initialize the empty logfile by writing a header marker
                int res = writeHeader(slot);
  
                if (res != TRI_ERROR_NO_ERROR) {
                  return MMFilesWalSlotInfo(res);
                }
  
                // new datafile. must write a prologue
                if (databaseId > 0 && collectionId > 0 && !mustWritePrologue) {
                  alignedSize = size + PrologueSize;
                  mustWritePrologue = true;
                }
  
                // advance to next slot
                slot = &_slots[_handoutIndex];
                _logfileManager->setLogfileOpen(_logfile);
              } else {
                TRI_ASSERT(status == MMFilesWalLogfile::StatusType::OPEN);
              }
            }
          }
  
          // if we get here, we got a free slot for the actual data...
  
          char* mem = _logfile->reserve(alignedSize);
  
          if (mem == nullptr) {
            LOG_TOPIC("cba38", WARN, arangodb::Logger::ENGINES)
                << "could not find free WAL slot";
            return MMFilesWalSlotInfo(TRI_ERROR_INTERNAL);
          }
  
          TRI_ASSERT(reinterpret_cast<uintptr_t>(mem) % 8 == 0);
  
          if (mustWritePrologue) {
            // write prologue...
  
            // hand out the prologue slot and directly fill it
            int res = writePrologue(slot, mem, databaseId, collectionId);
  
            if (res != TRI_ERROR_NO_ERROR) {
              return MMFilesWalSlotInfo(res);
            }
  
            // now return the slot
            mem += PrologueSize;  // advance memory pointer
            TRI_ASSERT(reinterpret_cast<uintptr_t>(mem) % 8 == 0);
  
            // use following slot for the actual data
            slot = &_slots[_handoutIndex];
  
            // note database and collection id for next time
            _lastDatabaseId = databaseId;
            _lastCollectionId = collectionId;
          }
  
          // only in this case we return a valid slot
          slot->setUsed(static_cast<void*>(mem), size, _logfile, handout());
  
          return MMFilesWalSlotInfo(slot);
        }
      }
  
      // if we get here, all slots are busy
      CONDITION_LOCKER(guard, _condition);
      if (!hasWaited) {
        ++_waiting;
        _logfileManager->signalSync(true);
        hasWaited = true;
      }
  
      bool mustWait;
      {
        MUTEX_LOCKER(mutexLocker, _lock);
        mustWait = (_freeSlots < 2);
      }
  
      if (mustWait) {
        guard.wait(10 * 1000);
      }
    }
  
    return MMFilesWalSlotInfo(TRI_ERROR_ARANGO_NO_JOURNAL);
  }
  
  /// @brief return a used slot, allowing its synchronization
  int MMFilesWalSlots::returnUsed(MMFilesWalSlotInfo& slotInfo, bool wakeUpSynchronizer,
                                  bool waitForSyncRequested, bool waitUntilSyncDone) {
    TRI_ASSERT(slotInfo.slot != nullptr);
    // waitUntilSyncDone does not make sense without waitForSyncRequested
    TRI_ASSERT(!waitUntilSyncDone || waitForSyncRequested);
  
    MMFilesWalSlot::TickType tick = slotInfo.slot->tick();
    MMFilesMarker const* marker =
        reinterpret_cast<MMFilesMarker const*>(slotInfo.slot->mem());
    MMFilesWalLogfile const* logfile = slotInfo.logfile;
  
    TRI_ASSERT(tick > 0);
  
    {
      MUTEX_LOCKER(mutexLocker, _lock);
      TRI_UpdateTicksDatafile(logfile->df(), marker);
      slotInfo.slot->setReturned(waitForSyncRequested);
      if (waitForSyncRequested) {
        ++_numEventsSync;
      } else {
        ++_numEvents;
      }
    }
  
    wakeUpSynchronizer |= waitForSyncRequested;
    wakeUpSynchronizer |= waitUntilSyncDone;
  
    if (wakeUpSynchronizer) {
      _logfileManager->signalSync(waitForSyncRequested);
    }
  
    if (waitUntilSyncDone) {
      // on shutdown, return early
      if (application_features::ApplicationServer::isStopping()) {
        return TRI_ERROR_SHUTTING_DOWN;
      }
  
      waitForTick(tick);
    }
  
    return TRI_ERROR_NO_ERROR;
  }
  
  /// @brief get the next synchronizable region
  MMFilesWalSyncRegion MMFilesWalSlots::getSyncRegion() {
    bool sealRequested = false;
    MMFilesWalSyncRegion region;
  
    MUTEX_LOCKER(mutexLocker, _lock);
  
    size_t slotIndex = _recycleIndex;
  
    while (true) {
      MMFilesWalSlot const* slot = &_slots[slotIndex];
      TRI_ASSERT(slot != nullptr);
  
      if (sealRequested && slot->isUnused()) {
        region.canSeal = true;
      }
  
      if (!slot->isReturned()) {
        // found a slot that is not yet returned
        // if it belongs to another logfile, we can seal the logfile we created
        // the region for
        auto otherId = slot->logfileId();
  
        if (region.logfileId != 0 && otherId != 0 && otherId != region.logfileId) {
          region.canSeal = true;
        }
        break;
      }
  
      if (region.logfileId == 0) {
        // first member
        MMFilesWalLogfile::StatusType status;
  
        region.logfileId = slot->logfileId();
        // the following call also updates status
        region.logfile = _logfileManager->getLogfile(slot->logfileId(), status);
        region.mem = static_cast<char*>(slot->mem());
        region.size = slot->size();
        region.logfileStatus = status;
        region.firstSlotIndex = slotIndex;
        region.lastSlotIndex = slotIndex;
        region.waitForSync = slot->waitForSync();
  
        if (status == MMFilesWalLogfile::StatusType::SEAL_REQUESTED) {
          sealRequested = true;
        }
      } else {
        if (slot->logfileId() != region.logfileId) {
          // got a different logfile
          region.checkMore = true;
          region.canSeal = true;
          break;
        }
  
        // this is a group commit!!
  
        // update the region
        region.size += (uint32_t)(static_cast<char*>(slot->mem()) -
                                  (region.mem + region.size) + slot->size());
        region.lastSlotIndex = slotIndex;
        region.waitForSync |= slot->waitForSync();
      }
  
      if (++slotIndex >= _numberOfSlots) {
        slotIndex = 0;
      }
  
      if (slotIndex == _recycleIndex) {
        // one full loop
        break;
      }
    }
  
    return region;
  }
  
  /// @brief return a region to the freelist
  void MMFilesWalSlots::returnSyncRegion(MMFilesWalSyncRegion const& region) {
    TRI_ASSERT(region.logfileId != 0);
  
    size_t slotIndex = region.firstSlotIndex;
  
    {
      MUTEX_LOCKER(mutexLocker, _lock);
  
      while (true) {
        MMFilesWalSlot* slot = &_slots[slotIndex];
        TRI_ASSERT(slot != nullptr);
  
        // note last tick
        MMFilesWalSlot::TickType tick = slot->tick();
        TRI_ASSERT(tick >= _lastCommittedTick);
        _lastCommittedTick = tick;
  
        // update the data tick
        MMFilesMarker const* m = static_cast<MMFilesMarker const*>(slot->mem());
        if (m->getType() != TRI_DF_MARKER_HEADER && m->getType() != TRI_DF_MARKER_FOOTER) {
          _lastCommittedDataTick = tick;
        }
  
        region.logfile->update(m);
  
        slot->setUnused();
        ++_freeSlots;
  
        // update recycle index, too
        if (++_recycleIndex >= _numberOfSlots) {
          _recycleIndex = 0;
        }
  
        if (slotIndex == region.lastSlotIndex) {
          break;
        }
  
        if (++slotIndex >= _numberOfSlots) {
          slotIndex = 0;
        }
      }
    }
  
    // signal that we have done something
    CONDITION_LOCKER(guard, _condition);
  
    if (_waiting > 0 || region.waitForSync) {
      _condition.broadcast();
    }
  }
  
  /// @brief get the current open region of a logfile
  /// this uses the slots lock
  void MMFilesWalSlots::getActiveLogfileRegion(MMFilesWalLogfile* logfile,
                                               char const*& begin, char const*& end) {
    MUTEX_LOCKER(mutexLocker, _lock);
  
    MMFilesDatafile* datafile = logfile->df();
  
    begin = datafile->_data;
    end = begin + datafile->currentSize();
  }
  
  /// @brief get the current tick range of a logfile
  /// this uses the slots lock
  void MMFilesWalSlots::getActiveTickRange(MMFilesWalLogfile* logfile,
                                           TRI_voc_tick_t& tickMin, TRI_voc_tick_t& tickMax) {
    MUTEX_LOCKER(mutexLocker, _lock);
  
    MMFilesDatafile* datafile = logfile->df();
  
    tickMin = datafile->_tickMin;
    tickMax = datafile->_tickMax;
  }
  
  /// @brief close a logfile
  int MMFilesWalSlots::closeLogfile(MMFilesWalSlot::TickType& lastCommittedTick, bool& worked) {
    bool hasWaited = false;
    worked = false;
  
    double const maxWait = 30.0;
    double const end = TRI_microtime() + maxWait;
    while (true) {
      {
        MUTEX_LOCKER(mutexLocker, _lock);
  
        lastCommittedTick = _lastCommittedTick;
  
        MMFilesWalSlot* slot = &_slots[_handoutIndex];
        TRI_ASSERT(slot != nullptr);
  
        if (slot->isUnused()) {
          if (hasWaited) {
            CONDITION_LOCKER(guard, _condition);
            TRI_ASSERT(_waiting > 0);
            --_waiting;
            hasWaited = false;
          }
  
          if (_logfile != nullptr) {
            if (_logfile->status() == MMFilesWalLogfile::StatusType::EMPTY) {
              // no need to seal a still-empty logfile
              return TRI_ERROR_NO_ERROR;
            }
  
            // seal existing logfile by creating a footer marker
            int res = writeFooter(slot);
  
            if (res != TRI_ERROR_NO_ERROR) {
              LOG_TOPIC("bc339", ERR, arangodb::Logger::ENGINES)
                  << "could not write logfile footer: " << TRI_errno_string(res);
              return res;
            }
  
            _logfileManager->setLogfileSealRequested(_logfile);
  
            // advance to next slot
            slot = &_slots[_handoutIndex];
  
            // invalidate the logfile so for the next write we'll use a
            // new one
            _logfile = nullptr;
  
            // intentionally falls through
          }
  
          TRI_IF_FAILURE("LogfileManagerGetWriteableLogfile") {
            return TRI_ERROR_ARANGO_NO_JOURNAL;
          }
  
          TRI_ASSERT(_logfile == nullptr);
          // fetch the next free logfile (this may create a new one)
          // note: as we don't have a real marker to write the size does
          // not matter (we use a size of 1 as it must be > 0)
          MMFilesWalLogfile::StatusType status;
          int res = newLogfile(1, status);
  
          if (res != TRI_ERROR_NO_ERROR) {
            if (res != TRI_ERROR_ARANGO_NO_JOURNAL) {
              return res;
            }
  
            std::this_thread::sleep_for(std::chrono::milliseconds(10));
            // try again in next iteration
          } else {
            TRI_ASSERT(_logfile != nullptr);
  
            if (status == MMFilesWalLogfile::StatusType::EMPTY) {
              // initialize the empty logfile by writing a header marker
              res = writeHeader(slot);
  
              if (res != TRI_ERROR_NO_ERROR) {
                LOG_TOPIC("9c7f8", ERR, arangodb::Logger::ENGINES)
                    << "could not write logfile header: " << TRI_errno_string(res);
                return res;
              }
  
              _logfileManager->setLogfileOpen(_logfile);
              worked = true;
            } else {
              TRI_ASSERT(status == MMFilesWalLogfile::StatusType::OPEN);
              worked = false;
            }
  
            return TRI_ERROR_NO_ERROR;
          }
        }
      }
  
      // if we get here, all slots are busy
      CONDITION_LOCKER(guard, _condition);
      if (!hasWaited) {
        ++_waiting;
        hasWaited = true;
      }
  
      bool mustWait;
      {
        MUTEX_LOCKER(mutexLocker, _lock);
        mustWait = (_freeSlots < 2);
      }
  
      if (mustWait) {
        guard.wait(10 * 1000);
      }
  
      if (TRI_microtime() >= end) {
        // time's up!
        break;
      }
    }
  
    return TRI_ERROR_ARANGO_NO_JOURNAL;
  }
  
  /// @brief write a header marker
  int MMFilesWalSlots::writeHeader(MMFilesWalSlot* slot) {
    TRI_ASSERT(_logfile != nullptr);
  
    MMFilesDatafileHeaderMarker header = MMFilesDatafileHelper::CreateHeaderMarker(
        static_cast<uint32_t>(_logfile->allocatedSize()),
        static_cast<TRI_voc_fid_t>(_logfile->id()));
    size_t const size = header.base.getSize();
  
    auto* mem = static_cast<void*>(_logfile->reserve(size));
    TRI_ASSERT(mem != nullptr);
    TRI_ASSERT(_logfile != nullptr);
  
    slot->setUsed(mem, static_cast<uint32_t>(size), _logfile, handout());
    slot->fill(&header.base, size);
    slot->setReturned(false);  // no sync
  
    // reset values for next write
    _lastDatabaseId = 0;
    _lastCollectionId = 0;
  
    return TRI_ERROR_NO_ERROR;
  }
  
  /// @brief write a prologue marker
  int MMFilesWalSlots::writePrologue(MMFilesWalSlot* slot, void* mem,
                                     TRI_voc_tick_t databaseId, TRI_voc_cid_t collectionId) {
    MMFilesPrologueMarker header =
        MMFilesDatafileHelper::CreatePrologueMarker(databaseId, collectionId);
    size_t const size = header.base.getSize();
  
    TRI_ASSERT(size == PrologueSize);
  
    TRI_ASSERT(mem != nullptr);
    TRI_ASSERT(_logfile != nullptr);
  
    slot->setUsed(mem, static_cast<uint32_t>(size), _logfile, handout());
    slot->fill(&header.base, size);
    slot->setReturned(false);  // no sync
  
    return TRI_ERROR_NO_ERROR;
  }
  
  /// @brief write a footer marker
  int MMFilesWalSlots::writeFooter(MMFilesWalSlot* slot) {
    TRI_ASSERT(_logfile != nullptr);
  
    MMFilesDatafileFooterMarker footer = MMFilesDatafileHelper::CreateFooterMarker();
    size_t const size = footer.base.getSize();
  
    auto* mem = static_cast<void*>(_logfile->reserve(size));
    TRI_ASSERT(mem != nullptr);
    TRI_ASSERT(_logfile != nullptr);
  
    slot->setUsed(mem, static_cast<uint32_t>(size), _logfile, handout());
    slot->fill(&footer.base, size);
    slot->setReturned(true);  // sync
  
    // reset values for next write
    _lastDatabaseId = 0;
    _lastCollectionId = 0;
  
    return TRI_ERROR_NO_ERROR;
  }
  
  /// @brief handout a region and advance the handout index
  MMFilesWalSlot::TickType MMFilesWalSlots::handout() {
    TRI_ASSERT(_freeSlots > 0);
    _freeSlots--;
  
    if (++_handoutIndex == _numberOfSlots) {
      // wrap around
      _handoutIndex = 0;
    }
  
    _lastAssignedTick = static_cast<MMFilesWalSlot::TickType>(TRI_NewTickServer());
    return _lastAssignedTick;
  }
  
  /// @brief return the next slots that would be handed out, without
  /// actually handing it out
  size_t MMFilesWalSlots::nextHandoutIndex() const {
    size_t handoutIndex = _handoutIndex;
    if (++handoutIndex == _numberOfSlots) {
      handoutIndex = 0;
    }
    return handoutIndex;
  }
  
  /// @brief wait until all data has been synced up to a certain marker
  bool MMFilesWalSlots::waitForTick(MMFilesWalSlot::TickType tick) {
    static uint64_t const SleepTime = 10000;
    static int const MaxIterations = 30 * 1000 * 1000 / SleepTime;
    int iterations = 0;
  
    // wait until data has been committed to disk
    do {
      if (lastCommittedTick() >= tick) {
        return true;
      }
  
      CONDITION_LOCKER(guard, _condition);
      guard.wait(SleepTime);
    } while (++iterations < MaxIterations);
  
    return false;
  }
  
  /// @brief request a new logfile which can satisfy a marker of the
  /// specified size
  int MMFilesWalSlots::newLogfile(uint32_t size, MMFilesWalLogfile::StatusType& status) {
    TRI_ASSERT(size > 0);
  
    if (_shutdown) {
      return TRI_ERROR_REQUEST_CANCELED;
    }
  
    status = MMFilesWalLogfile::StatusType::UNKNOWN;
    MMFilesWalLogfile* logfile = nullptr;
    int res = _logfileManager->getWriteableLogfile(size, status, logfile);
  
    if (res == TRI_ERROR_NO_ERROR) {
      TRI_ASSERT(logfile != nullptr);
      _logfile = logfile;
    } else if (res == TRI_ERROR_LOCK_TIMEOUT) {
      _logfileManager->logStatus();
    }
  
    return res;
  }
  