<commit_before>//---------------------------------------------------------- -*- Mode: C++ -*-
// $Id$
//
// Created 2010/06/11
// Author: <NAME>
//
// Copyright 2010-2012 Quantcast Corp.
//
// This file is part of Kosmos File System (KFS).
//
// 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.
//
//
//----------------------------------------------------------------------------

#include "Writer.h"

#include <sstream>
#include <algorithm>
#include <cerrno>
#include <sstream>
#include <bitset>
#include <string.h>

#include "kfsio/IOBuffer.h"
#include "kfsio/NetManager.h"
#include "kfsio/checksum.h"
#include "kfsio/ITimeout.h"
#include "kfsio/ClientAuthContext.h"
#include "common/kfsdecls.h"
#include "common/MsgLogger.h"
#include "qcdio/QCUtils.h"
#include "qcdio/qcstutils.h"
#include "qcdio/qcdebug.h"
#include "qcdio/QCDLList.h"
#include "RSStriper.h"
#include "KfsOps.h"
#include "utils.h"
#include "KfsClient.h"

namespace KFS
{
namespace client
{

using std::min;
using std::max;
using std::string;
using std::ostream;
using std::ostringstream;
using std::istringstream;

// Kfs client write state machine implementation.
class Writer::Impl :
    public QCRefCountedObj,
    private ITimeout,
    private KfsNetClient::OpOwner
{
public:
    typedef QCRefCountedObj::StRef StRef;

    enum
    {
        kErrorNone       = 0,
        kErrorParameters = -EINVAL,
        kErrorTryAgain   = -EAGAIN,
        kErrorFault      = -EFAULT,
        kErrorNoEntry    = -ENOENT
    };

    Impl(
        Writer&     inOuter,
        MetaServer& inMetaServer,
        Completion* inCompletionPtr,
        int         inMaxRetryCount,
        int         inWriteThreshold,
        int         inMaxPartialBuffersCount,
        int         inTimeSecBetweenRetries,
        int         inOpTimeoutSec,
        int         inIdleTimeoutSec,
        int         inMaxWriteSize,
        string      inLogPrefix,
        int64_t     inChunkServerInitialSeqNum)
        : QCRefCountedObj(),
          ITimeout(),
          KfsNetClient::OpOwner(),
          mOuter(inOuter),
          mMetaServer(inMetaServer),
          mPathName(),
          mFileId(-1),
          mClosingFlag(false),
          mSleepingFlag(false),
          mErrorCode(0),
          mWriteThreshold(max(0, inWriteThreshold)),
          mPartialBuffersCount(0),
          mPendingCount(0),
          mIdleTimeoutSec(inIdleTimeoutSec),
          mOpTimeoutSec(inOpTimeoutSec),
          mMaxRetryCount(inMaxRetryCount),
          mTimeSecBetweenRetries(inTimeSecBetweenRetries),
          mMaxPartialBuffersCount(inMaxPartialBuffersCount),
          mMaxWriteSize(min((int)CHUNKSIZE,
            (int)((max(0, inMaxWriteSize) + CHECKSUM_BLOCKSIZE - 1) /
                CHECKSUM_BLOCKSIZE * CHECKSUM_BLOCKSIZE))),
          mReplicaCount(-1),
          mRetryCount(0),
          mFileSize(0),
          mOffset(0),
          mOpenChunkBlockSize(CHUNKSIZE),
          mChunkServerInitialSeqNum(inChunkServerInitialSeqNum),
          mCompletionPtr(inCompletionPtr),
          mBuffer(),
          mLogPrefix(inLogPrefix),
          mStats(),
          mNetManager(mMetaServer.GetNetManager()),
          mTruncateOp(0, 0, -1, 0),
          mOpStartTime(0),
          mCompletionDepthCount(0),
          mStriperProcessCount(0),
          mStriperPtr(0)
        { Writers::Init(mWriters); }
    int Open(
        kfsFileId_t inFileId,
        const char* inFileNamePtr,
        Offset      inFileSize,
        int         inStriperType,
        int         inStripeSize,
        int         inStripeCount,
        int         inRecoveryStripeCount,
        int         inReplicaCount)
    {
        if (inFileId <= 0 || ! inFileNamePtr || ! *inFileNamePtr) {
            return kErrorParameters;
        }
        if (mFileId > 0) {
            if (inFileId == mFileId &&
                    inFileNamePtr == mPathName) {
                return mErrorCode;
            }
            return kErrorParameters;
        }
        if (IsOpen() && mErrorCode != 0) {
            return (mErrorCode < 0 ? mErrorCode : -mErrorCode);
        }
        if (mClosingFlag || mSleepingFlag) {
            return kErrorTryAgain;
        }
        delete mStriperPtr;
        string theErrMsg;
        mStriperPtr = 0;
        mOpenChunkBlockSize = Offset(CHUNKSIZE);
        mStriperPtr = Striper::Create(
            inStriperType,
            inStripeCount,
            inRecoveryStripeCount,
            inStripeSize,
            inFileSize,
            mLogPrefix,
            *this,
            mOpenChunkBlockSize,
            theErrMsg
        );
        if (! theErrMsg.empty()) {
            KFS_LOG_STREAM_ERROR << mLogPrefix <<
                theErrMsg <<
            KFS_LOG_EOM;
            return kErrorParameters;
        }
        if (! mStriperPtr || mOpenChunkBlockSize < Offset(CHUNKSIZE)) {
            mOpenChunkBlockSize = Offset(CHUNKSIZE);
        }
        mBuffer.Clear();
        mStats.Clear();
        mReplicaCount          = inReplicaCount;
        mFileSize              = inFileSize;
        mPartialBuffersCount   = 0;
        mPathName              = inFileNamePtr;
        mErrorCode             = 0;
        mFileId                = inFileId;
        mTruncateOp.fid        = -1;
        mTruncateOp.pathname   = 0;
        mTruncateOp.fileOffset = mFileSize;
        mRetryCount            = 0;
        return StartWrite();
    }
    int Close()
    {
        if (! IsOpen()) {
            return 0;
        }
        if (mErrorCode != 0) {
            return mErrorCode;
        }
        if (mClosingFlag) {
            return kErrorTryAgain;
        }
        mClosingFlag = true;
        return StartWrite();
    }
    int Write(
        IOBuffer& inBuffer,
        int       inLength,
        Offset    inOffset,
        bool      inFlushFlag,
        int       inWriteThreshold)
    {
        if (inOffset < 0) {
            return kErrorParameters;
        }
        if (mErrorCode != 0) {
            return (mErrorCode < 0 ? mErrorCode : -mErrorCode);
        }
        if (mClosingFlag || ! IsOpen()) {
            return kErrorParameters;
        }
        if (inLength <= 0) {
            return (
                (ReportCompletion(0, inLength, inOffset) && inFlushFlag) ?
                StartWrite(true) : 0
            );
        }
        if (inOffset != mOffset + mBuffer.BytesConsumable()) {
            // Just flush for now, do not try to optimize buffer rewrite.
            const int thePrevRefCount = GetRefCount();
            const int theRet = Flush();
            if (theRet < 0) {
                return theRet;
            }
            if (thePrevRefCount > GetRefCount()) {
                return (mErrorCode < 0 ? mErrorCode : -mErrorCode);
            }
            mOffset = inOffset;
        }
        if (mMaxPartialBuffersCount == 0 ||
                inLength < IOBufferData::GetDefaultBufferSize() * 2) {
            // If write size is small, then copy it into the last buffer.
            mBuffer.ReplaceKeepBuffersFull(
                &inBuffer, mBuffer.BytesConsumable(), inLength);
        } else {
            if (mBuffer.IsEmpty()) {
                mPartialBuffersCount = 0;
            }
            mBuffer.Move(&inBuffer, inLength);
            mPartialBuffersCount++;
            if (mMaxPartialBuffersCount >= 0 &&
                    mPartialBuffersCount >= mMaxPartialBuffersCount) {
                mBuffer.MakeBuffersFull();
                mPartialBuffersCount = 0;
                mStats.mBufferCompactionCount++;
            }
        }
        if (inWriteThreshold >= 0) {
            mWriteThreshold = inWriteThreshold;
        }
        const int theErrorCode = StartWrite(inFlushFlag);
        return (theErrorCode == 0 ? inLength :
            (theErrorCode < 0 ? theErrorCode : -theErrorCode));
    }
    int Flush()
    {
        const int theErrorCode = StartWrite(true);
        return (theErrorCode < 0 ? theErrorCode : -theErrorCode);
    }
    void Stop()
    {
        while (! Writers::IsEmpty(mWriters)) {
            delete Writers::Front(mWriters);
        }
        if (mTruncateOp.fid >= 0) {
            mMetaServer.Cancel(&mTruncateOp, this);
        }
        if (mSleepingFlag) {
            mNetManager.UnRegisterTimeoutHandler(this);
            mSleepingFlag = false;
        }
        mClosingFlag = false;
        mBuffer.Clear();
    }
    void Shutdown()
    {
        Stop();
        mFileId      = -1;
        mErrorCode   = 0;
    }
    bool IsOpen() const
        { return (mFileId > 0); }
    bool IsClosing() const
        { return (IsOpen() && mClosingFlag); }
    bool IsActive() const
    {
        return (
            IsOpen() && (
                ! mBuffer.IsEmpty() ||
                ! Writers::IsEmpty(mWriters) ||
                mClosingFlag)
        );
    }
    Offset GetPendingSize() const
        { return (GetPendingSizeSelf() + mPendingCount); }
    int SetWriteThreshold(
        int inThreshold)
    {
        const int  theThreshold      = max(0, inThreshold);
        const bool theStartWriteFlag = mWriteThreshold > theThreshold;
        mWriteThreshold = theThreshold;
        return ((theStartWriteFlag && IsOpen() && mErrorCode == 0) ?
            StartWrite() : mErrorCode
        );
    }
    void DisableCompletion()
        { mCompletionPtr = 0; }
    void Register(
        Completion* inCompletionPtr)
    {
        if (inCompletionPtr == mCompletionPtr) {
            return;
        }
        if (mCompletionPtr) {
            mCompletionPtr->Unregistered(mOuter);
        }
        mCompletionPtr = inCompletionPtr;
    }
    bool Unregister(
        Completion* inCompletionPtr)
    {
        if (inCompletionPtr != mCompletionPtr) {
            return false;
        }
        mCompletionPtr = 0;
        return true;
    }
    void GetStats(
        Stats&               outStats,
        KfsNetClient::Stats& outChunkServersStats)
    {
        outStats             = mStats;
        outChunkServersStats = mChunkServersStats;
    }
    bool GetErrorCode() const
        { return mErrorCode; }

private:
    typedef KfsNetClient ChunkServer;

    class ChunkWriter : private ITimeout, private KfsNetClient::OpOwner
    {
    public:
        struct WriteOp;
        typedef QCDLList<WriteOp, 0> Queue;
        typedef QCDLList<ChunkWriter, 0> Writers;

        struct WriteOp : public KfsOp
        {
            WritePrepareOp mWritePrepareOp;
            WriteSyncOp    mWriteSyncOp;
            IOBuffer       mBuffer;
            size_t         mBeginBlock;
            size_t         mEndBlock;
            time_t         mOpStartTime;
            bool           mChecksumValidFlag;
            WriteOp*       mPrevPtr[1];
            WriteOp*       mNextPtr[1];

            WriteOp()
                : KfsOp(CMD_WRITE, 0),
                  mWritePrepareOp(0, 0, 0),
                  mWriteSyncOp(),
                  mBuffer(),
                  mBeginBlock(0),
                  mEndBlock(0),
                  mOpStartTime(0),
                  mChecksumValidFlag(false)
                { Queue::Init(*this); }
            void Delete(
                WriteOp** inListPtr)
            {
                Queue::Remove(inListPtr, *this);
                delete this;
            }
            virtual void Request(
                ostream& inStream)
            {
                if (mWritePrepareOp.replyRequestedFlag) {
                    mWritePrepareOp.seq = seq;
                } else {
                    mWriteSyncOp.seq    = seq;
                    mWritePrepareOp.seq = seq + 1;
                }
                mWritePrepareOp.Request(inStream);
            }
            virtual bool NextRequest(
                kfsSeq_t inSeqNum,
                ostream& inStream)
            {
                if (mWritePrepareOp.replyRequestedFlag) {
                    return false;
                }
                QCASSERT(seq <= inSeqNum && inSeqNum <= mWritePrepareOp.seq + 1);
                if (mWritePrepareOp.seq < inSeqNum) {
                    return false;
                }
                mWriteSyncOp.Request(inStream);
                return true;
            }
            virtual ostream& ShowSelf(
                ostream& inStream) const
            {
                inStream << mWritePrepareOp.Show();
                if (! mWritePrepareOp.replyRequestedFlag) {
                    inStream << " " << mWriteSyncOp.Show();
                }
                return inStream;
            }
            virtual void ParseResponseHeaderSelf(
                const Properties& inProps)
            {
                if (contentLength > 0) {
                    KFS_LOG_STREAM_ERROR <<
                        "invalid response content length: " << contentLength <<
                        " " << mWriteSyncOp.Show() <<
                    KFS_LOG_EOM;
                    contentLength = 0;
                }
                mWritePrepareOp.status    = status;
                mWritePrepareOp.statusMsg = statusMsg;
                mWriteSyncOp.status       = status;
                mWriteSyncOp.statusMsg    = statusMsg;
                if (mWritePrepareOp.replyRequestedFlag) {
                    mWritePrepareOp.ParseResponseHeaderSelf(inProps);
                } else {
                    mWriteSyncOp.ParseResponseHeaderSelf(inProps);
                }
            }
            void InitBlockRange()
            {
                QCASSERT(
                    mWritePrepareOp.offset >= 0 &&
                    mWritePrepareOp.offset +
                        mBuffer.BytesConsumable() <= (Offset)CHUNKSIZE
                );
                mBeginBlock = mWritePrepareOp.offset / CHECKSUM_BLOCKSIZE;
                mEndBlock   = mBeginBlock +
                    (mBuffer.BytesConsumable() + CHECKSUM_BLOCKSIZE - 1) /
                    CHECKSUM_BLOCKSIZE;
            }
        private:
            virtual ~WriteOp()
                {}
            WriteOp(
                const WriteOp& inWriteOp);
            WriteOp& operator=(
                const WriteOp& inWriteOp);
        };

        ChunkWriter(
            Impl&         inOuter,
            int64_t       inSeqNum,
            const string& inLogPrefix)
            : ITimeout(),
              KfsNetClient::OpOwner(),
              mOuter(inOuter),
              mChunkServer(
                inOuter.mNetManager,
                string(), -1,
                // All chunk server retries are handled here
                0, // inMaxRetryCount
                0, // inTimeSecBetweenRetries,
                inOuter.mOpTimeoutSec,
                inOuter.mIdleTimeoutSec,
                inSeqNum,
                inLogPrefix.c_str(),
                // Just fail the op. Error handler will reset connection and
                // cancel all pending ops by calling Stop()
                false // inResetConnectionOnOpTimeoutFlag
              ),
              mErrorCode(0),
              mRetryCount(0),
              mPendingCount(0),
              mOpenChunkBlockFileOffset(-1),
              mOpStartTime(0),
              mWriteIds(),
              mAllocOp(0, 0, ""),
              mWriteIdAllocOp(0, 0, 0, 0, 0),
              mCloseOp(0, 0),
              mLastOpPtr(0),
              mSleepingFlag(false),
              mClosingFlag(false),
              mLogPrefix(inLogPrefix),
              mOpDoneFlagPtr(0),
              mInFlightBlocks(),
              mHasSubjectIdFlag(),
              mChunkAccess(),
              mLeaseExpireTime(0),
              mChunkAccessExpireTime(0),
              mCSAccessExpireTime(0)
        {
            Queue::Init(mPendingQueue);
            Queue::Init(mInFlightQueue);
            Writers::Init(*this);
            Writers::PushFront(mOuter.mWriters, *this);
            mChunkServer.SetRetryConnectOnly(true);
            mAllocOp.fileOffset        = -1;
            mAllocOp.invalidateAllFlag = false;
        }
        ~ChunkWriter()
        {
            ChunkWriter::Shutdown();
            ChunkServer::Stats theStats;
            mChunkServer.GetStats(theStats);
            mOuter.mChunkServersStats.Add(theStats);
            Writers::Remove(mOuter.mWriters, *this);
        }
        void CancelClose()
            { mClosingFlag = false; }
        // The QueueWrite() guarantees that completion will not be invoked.
        // The writes will be queued even if the writer is already in the error
        // state: mErrorCode != 0. In the case of fatal error all pending writes
        // are discarded when the writer gets deleted.
        //
        // StartWrite() must be called in order to start executing pending
        // writes.
        // This allows the caller to properly update its state before the writes
        // get executed, and the corresponding completion(s) invoked.
        int QueueWrite(
            IOBuffer& inBuffer,
            int       inSize,
            Offset    inOffset,
            int       inWriteThreshold)
        {
            int theSize = min(inBuffer.BytesConsumable(), inSize);
            if (theSize <= 0) {
                return 0;
            }
            const Offset kChunkSize         = (Offset)CHUNKSIZE;
            const int    kChecksumBlockSize = (int)CHECKSUM_BLOCKSIZE;
            QCRTASSERT(inOffset >= 0 && ! mClosingFlag);
            const Offset theChunkOffset = inOffset % kChunkSize;
            if (mAllocOp.fileOffset < 0) {
                mAllocOp.fileOffset = inOffset - theChunkOffset;
                mOpenChunkBlockFileOffset = mAllocOp.fileOffset -
                    mAllocOp.fileOffset % mOuter.mOpenChunkBlockSize;
            } else {
                QCRTASSERT(mAllocOp.fileOffset == inOffset - theChunkOffset);
            }
            theSize = min(theSize, (int)(kChunkSize - theChunkOffset));
            QCASSERT(theSize > 0);
            Offset thePos = theChunkOffset;
            // Try to append to the last pending op.
            WriteOp* const theLastOpPtr = Queue::Back(mPendingQueue);
            if (theLastOpPtr) {
                WriteOp& theOp = *theLastOpPtr;
                const int    theOpSize = theOp.mBuffer.BytesConsumable();
                const Offset theOpPos  = theOp.mWritePrepareOp.offset;
                if (theOpPos + theOpSize == thePos) {
                    const int theHead = (int)(theOpPos % kChecksumBlockSize);
                    int       theNWr  = min(theSize,
                        (theHead == 0 ?
                            mOuter.mMaxWriteSize :
                            kChecksumBlockSize - theHead
                        ) - theOpSize
                    );
                    if (theNWr > 0 &&
                            theOpSize + theNWr > kChecksumBlockSize) {
                        theNWr -= (theOpSize + theNWr) % kChecksumBlockSize;
                    }
                    if (theNWr > 0) {
                        theOp.mBuffer.Move(&inBuffer, theNWr);
                        // Force checksum recomputation.
                        theOp.mChecksumValidFlag = false;
                        theOp.mWritePrepareOp.checksums.clear();
                        // Update last the block index.
                        const int theCurBegin = theOp.mBeginBlock;
                        theOp.InitBlockRange();
                        theOp.mBeginBlock = theCurBegin;
                        // The op is already in the pending queue.
                        theSize -= theNWr;
                        thePos  += theNWr;
                    }
                }
            }
            const int theWriteThreshold = thePos + theSize >= kChunkSize ?
                1 : max(inWriteThreshold, 1);
            const int theBlockOff       = (int)(thePos % kChecksumBlockSize);
            if (theBlockOff > 0 && (theSize >= theWriteThreshold ||
                    theBlockOff + theSize >= kChecksumBlockSize)) {
                WriteOp* const theWriteOpPtr = new WriteOp();
                theWriteOpPtr->mWritePrepareOp.offset = thePos;
                const int theNWr = theWriteOpPtr->mBuffer.Move(
                    &inBuffer,
                    min(theSize, kChecksumBlockSize - theBlockOff)
                );
                theSize -= theNWr;
                thePos  += theNWr;
                theWriteOpPtr->InitBlockRange();
                Queue::PushBack(mPendingQueue, *theWriteOpPtr);
            }
            while (theSize >= theWriteThreshold) {
                int theOpSize = min(mOuter.mMaxWriteSize, theSize);
                if (theOpSize > kChecksumBlockSize) {
                    theOpSize -= theOpSize % kChecksumBlockSize;
                }
                WriteOp* const theWriteOpPtr = new WriteOp();
                theWriteOpPtr->mWritePrepareOp.offset = thePos;
                const int theNWr =
                    theWriteOpPtr->mBuffer.Move(&inBuffer, theOpSize);
                theSize -= theNWr;
                thePos  += theNWr;
                theWriteOpPtr->InitBlockRange();
                Queue::PushBack(mPendingQueue, *theWriteOpPtr);
            }
            QCRTASSERT(thePos <= kChunkSize && theSize >= 0);
            const Offset theNWr = thePos - theChunkOffset;
            // The following must be updated before invoking StartWrite(),
            // as it could invoke completion immediately (in the case of
            // failure).
            mPendingCount += theNWr;
            return theNWr;
        }
        void StartWrite()
        {
            if (mSleepingFlag) {
                return;
            }
            if (mErrorCode != 0 && ! mAllocOp.invalidateAllFlag) {
                if (mLastOpPtr) {
                    Reset();
                }
                mClosingFlag = false;
                return;
            }
            if (mClosingFlag && ! CanWrite()) {
                if (! Queue::IsEmpty(mInFlightQueue)) {
                    return;
                }
                if (mLastOpPtr == &mCloseOp) {
                    return;
                }
                // Try to close chunk even if chunk server disconnected, to
                // release the write lease.
                if (mAllocOp.chunkId > 0) {
                    CloseChunk();
                    return;
                }
                mChunkServer.Stop();
                if (mLastOpPtr == &mAllocOp) {
                    mOuter.mMetaServer.Cancel(mLastOpPtr, this);
                }
                mClosingFlag        = false;
                mAllocOp.fileOffset = -1;
                mAllocOp.chunkId    = -1;
                ReportCompletion();
                return;
            }
            if (! CanWrite()) {
                return;
            }
            if (0 < mAllocOp.chunkId && mLeaseExpireTime < Now()) {
                // When chunk server disconnects it might clean up write lease.
                // Start from the beginning -- chunk allocation.
                KFS_LOG_STREAM_DEBUG << mLogPrefix <<
                    "write lease expired: " <<
                        mChunkServer.GetServerLocation() <<
                    " starting from chunk allocation, pending:" <<
                    " queue: " << (Queue::IsEmpty(mPendingQueue) ? "" : "not") <<
                        " empty" <<
                KFS_LOG_EOM;
                Reset();
                if (! CanWrite()) {
                    // Do not try to preallocate chunk after inactivity timeout
                    // or error, if no data pending.
                    return;
                }
            }
            // Return immediately after calling Write() and AllocateChunk(), as
            // these can invoke completion. Completion, in turn, can delete
            // this.
            // Other methods of this class have to return immediately (unwind)
            // after invoking StartWrite().
            if (mAllocOp.chunkId > 0 && ! mWriteIds.empty()) {
                Write();
            } else if (! mLastOpPtr) { // Close can be in flight.
                Reset();
                AllocateChunk();
            }
        }
        void Close()
        {
            if (! mClosingFlag && IsOpen()) {
                mClosingFlag = true;
                StartWrite();
            }
        }
        void Shutdown()
        {
            Reset();
            QCRTASSERT(Queue::IsEmpty(mInFlightQueue));
            while (! Queue::IsEmpty(mPendingQueue)) {
                Queue::Front(mPendingQueue)->Delete(mPendingQueue);
            }
            mClosingFlag  = false;
            mErrorCode    = 0;
            mPendingCount = 0;
        }
        Offset GetFileOffset() const
        {
            return (mErrorCode == 0 ? mAllocOp.fileOffset : -1);
        }
        bool IsIdle() const
        {
            return (
                Queue::IsEmpty(mPendingQueue) &&
                Queue::IsEmpty(mInFlightQueue) &&
                ! mClosingFlag
            );
        }
        bool IsOpen() const
        {
            return (
                mErrorCode == 0 &&
                mAllocOp.fileOffset >= 0 &&
                ! mClosingFlag
            );
        }
        int GetErrorCode() const
            { return mErrorCode; }
        Offset GetPendingCount() const
            { return mPendingCount; }
        ChunkWriter* GetPrevPtr()
        {
            ChunkWriter& thePrev = ChunkWritersListOp::GetPrev(*this);
            return (&thePrev == this ? 0 : &thePrev);
        }
        Offset GetOpenChunkBlockFileOffset() const
        {
            return (mAllocOp.fileOffset >= 0 ? mOpenChunkBlockFileOffset : -1);
        }

    private:
        typedef std::vector<WriteInfo> WriteIds;
        typedef std::bitset<CHUNKSIZE / CHECKSUM_BLOCKSIZE> ChecksumBlocks;

        Impl&          mOuter;
        ChunkServer    mChunkServer;
        int            mErrorCode;
        int            mRetryCount;
        Offset         mPendingCount;
        Offset         mOpenChunkBlockFileOffset;
        time_t         mOpStartTime;
        WriteIds       mWriteIds;
        AllocateOp     mAllocOp;
        WriteIdAllocOp mWriteIdAllocOp;
        CloseOp        mCloseOp;
        KfsOp*         mLastOpPtr;
        bool           mSleepingFlag;
        bool           mClosingFlag;
        string const   mLogPrefix;
        bool*          mOpDoneFlagPtr;
        ChecksumBlocks mInFlightBlocks;
        bool           mHasSubjectIdFlag;
        string         mChunkAccess;
        time_t         mLeaseExpireTime;
        time_t         mChunkAccessExpireTime;
        time_t         mCSAccessExpireTime;
        WriteOp*       mPendingQueue[1];
        WriteOp*       mInFlightQueue[1];
        ChunkWriter*   mPrevPtr[1];
        ChunkWriter*   mNextPtr[1];

        friend class QCDLListOp<ChunkWriter, 0>;
        typedef QCDLListOp<ChunkWriter, 0> ChunkWritersListOp;

        void AllocateChunk()
        {
            QCASSERT(
                mOuter.mFileId > 0 &&
                mAllocOp.fileOffset >= 0 &&
                ! Queue::IsEmpty(mPendingQueue)
            );
            Reset(mAllocOp);
            mAllocOp.fid                  = mOuter.mFileId;
            mAllocOp.pathname             = mOuter.mPathName;
            mAllocOp.append               = false;
            mAllocOp.chunkId              = -1;
            mAllocOp.chunkVersion         = -1;
            mAllocOp.spaceReservationSize = 0;
            mAllocOp.maxAppendersPerChunk = 0;
            mAllocOp.allowCSClearTextFlag = false;
            mAllocOp.chunkServerAccessValidForTime = 0;
            mAllocOp.chunkServerAccessIssuedTime   = 0;
            mAllocOp.chunkServers.clear();
            mAllocOp.chunkAccess.clear();
            mAllocOp.chunkServerAccessToken.clear();
            mOuter.mStats.mChunkAllocCount++;
            EnqueueMeta(mAllocOp);
        }
        void Done(
            AllocateOp& inOp,
            bool        inCanceledFlag,
            IOBuffer*   inBufferPtr)
        {
            QCASSERT(&mAllocOp == &inOp && ! inBufferPtr);
            if (inCanceledFlag) {
                return;
            }
            if (inOp.status != 0 || (mAllocOp.chunkServers.empty() &&
                    ! mAllocOp.invalidateAllFlag)) {
                mAllocOp.chunkId = 0;
                HandleError(inOp);
                return;
            }
            if (mAllocOp.invalidateAllFlag) {
                // Report all writes completed. Completion does not expect the
                // offset to match the original write offset with striper.
                KFS_LOG_STREAM_INFO << mLogPrefix <<
                    "invalidate done:"
                    " chunk: "   << mAllocOp.chunkId <<
                    " offset: "  << mAllocOp.fileOffset <<
                    " status: "  << inOp.status <<
                    " pending: " << mPendingCount <<
                    " w-empty: " << Queue::IsEmpty(mPendingQueue) <<
                KFS_LOG_EOM;
                const int    theSize   = mPendingCount;
                const Offset theOffset = theSize > 0 ? mAllocOp.fileOffset : 0;
                mAllocOp.invalidateAllFlag = false;
                Shutdown();
                ReportCompletion(theOffset, theSize);
                return;
            }
            mLeaseExpireTime = Now() + LEASE_INTERVAL_SECS - 5;
            AllocateWriteId();
        }
        bool CanWrite()
        {
            return (
                ! Queue::IsEmpty(mPendingQueue) ||
                mAllocOp.invalidateAllFlag
            );
        }
        void AllocateWriteId()
        {
            QCASSERT(mAllocOp.chunkId > 0 && ! mAllocOp.chunkServers.empty());
            Reset(mWriteIdAllocOp);
            mWriteIdAllocOp.chunkId                     = mAllocOp.chunkId;
            mWriteIdAllocOp.chunkVersion                = mAllocOp.chunkVersion;
            mWriteIdAllocOp.isForRecordAppend           = false;
            mWriteIdAllocOp.chunkServerLoc              = mAllocOp.chunkServers;
            mWriteIdAllocOp.offset                      = 0;
            mWriteIdAllocOp.numBytes                    = 0;
            mWriteIdAllocOp.writePrepReplySupportedFlag = false;

            const time_t theNow = Now();
            mHasSubjectIdFlag = false;
            mChunkAccess.clear();

            const bool theCSClearTextAllowedFlag =
                mOuter.IsChunkServerClearTextAllowed();
            mChunkServer.SetShutdownSsl(
                mAllocOp.allowCSClearTextFlag &&
                theCSClearTextAllowedFlag
            );
            if (mAllocOp.chunkServerAccessToken.empty() ||
                    mAllocOp.chunkAccess.empty()) {
                mChunkServer.SetKey(0, 0, 0, 0);
                mChunkServer.SetAuthContext(0);
                if (! mAllocOp.chunkServerAccessToken.empty()) {
                    mWriteIdAllocOp.status    = -EINVAL;
                    mWriteIdAllocOp.statusMsg = "no chunk access";
                } else if (! mAllocOp.chunkAccess.empty()) {
                    mWriteIdAllocOp.status    = -EINVAL;
                    mWriteIdAllocOp.statusMsg = "no chunk server access";
                } else if (! theCSClearTextAllowedFlag) {
                    mWriteIdAllocOp.status    = -EPERM;
                    mWriteIdAllocOp.statusMsg =
                        "no cleart text chunk server access";
                } else {
                    mChunkAccessExpireTime = theNow + 60 * 60 * 24 * 365;
                    mCSAccessExpireTime    = mChunkAccessExpireTime;
                }
            } else {
                mChunkServer.SetKey(
                    mAllocOp.chunkServerAccessToken.data(),
                    mAllocOp.chunkServerAccessToken.size(),
                    mAllocOp.chunkServerAccessKey.GetPtr(),
                    mAllocOp.chunkServerAccessKey.GetSize()
                );
                mChunkAccess           = mAllocOp.chunkAccess;
                mWriteIdAllocOp.access = mChunkAccess;
                // Always ask for chunk access token here, as the chunk access
                // token's lifetime returned by alloc is 5 min.
                // The chunk returns the token with the corresponding key's
                // lifetime as the token subject includes write id.
                mWriteIdAllocOp.createChunkAccessFlag = true;
                mChunkAccessExpireTime = theNow - 60 * 60 * 24;
                mCSAccessExpireTime = GetAccessExpireTime(
                    theNow,
                    mAllocOp.chunkServerAccessIssuedTime,
                    mAllocOp.chunkServerAccessValidForTime
                );
                mWriteIdAllocOp.createChunkServerAccessFlag =
                    mCSAccessExpireTime <= theNow;
                if (mAllocOp.allowCSClearTextFlag &&
                        theCSClearTextAllowedFlag &&
                        mWriteIdAllocOp.createChunkServerAccessFlag) {
                    mWriteIdAllocOp.decryptKey = &mChunkServer.GetSessionKey();
                }
                if (! mChunkServer.GetAuthContext()) {
                    mChunkServer.SetAuthContext(
                        mOuter.mMetaServer.GetAuthContext());
                }
            }
            if (mWriteIdAllocOp.status == 0) {
                const bool kCancelPendingOpsFlag = true;
                if (mChunkServer.SetServer(
                        mAllocOp.chunkServers[0],
                        kCancelPendingOpsFlag,
                        &mWriteIdAllocOp.statusMsg)) {
                    Enqueue(mWriteIdAllocOp);
                    return;
                }
                mWriteIdAllocOp.status = kErrorFault;
            }
            HandleError(mWriteIdAllocOp);
        }
        static int64_t GetAccessExpireTime(
            time_t  inNow,
            int64_t inIssedTime,
            int64_t inValidFor)
        {
            // Use current time if the clock difference is large enough.
            int64_t theDiff = inIssedTime - (int64_t)inNow;
            if (theDiff < 0) {
                theDiff = -theDiff;
            }
            return (
                ((LEASE_INTERVAL_SECS * 3 < theDiff) ? inNow : inIssedTime) +
                inValidFor - LEASE_INTERVAL_SECS
            );
        }
        void UpdateAccess(
            ChunkAccessOp& inOp)
        {
            if (! inOp.chunkAccessResponse.empty()) {
                mHasSubjectIdFlag      = true;
                mChunkAccess           = inOp.chunkAccessResponse;
                mChunkAccessExpireTime = GetAccessExpireTime(
                    Now(),
                    inOp.accessResponseIssued,
                    inOp.accessResponseValidForSec
                );
            }
            if (0 < inOp.accessResponseValidForSec &&
                    ! inOp.chunkServerAccessId.empty()) {
                mChunkServer.SetKey(
                    inOp.chunkServerAccessId.data(),
                    inOp.chunkServerAccessId.size(),
                    inOp.chunkServerAccessKey.GetPtr(),
                    inOp.chunkServerAccessKey.GetSize()
                );
                if (inOp.chunkAccessResponse.empty()) {
                    mCSAccessExpireTime = GetAccessExpireTime(
                        Now(),
                        inOp.accessResponseIssued,
                        inOp.accessResponseValidForSec
                    );
                } else {
                    mCSAccessExpireTime = mChunkAccessExpireTime;
                }
            }
        }
        void SetAccess(
            ChunkAccessOp& inOp,
            bool           inCanRequestAccessFlag = true)
        {
            const time_t theNow = Now();
            inOp.access                      = mChunkAccess;
            inOp.createChunkAccessFlag       = inCanRequestAccessFlag &&
                mChunkAccessExpireTime <= theNow;
            inOp.createChunkServerAccessFlag = inCanRequestAccessFlag &&
                mCSAccessExpireTime    <= theNow;
            inOp.hasSubjectIdFlag            =
                mHasSubjectIdFlag && ! mWriteIds.empty();
            if (inOp.hasSubjectIdFlag) {
                inOp.subjectId = mWriteIds.front().writeId;
            }
            if (inOp.createChunkServerAccessFlag &&
                    mChunkServer.IsShutdownSsl()) {
                inOp.decryptKey = &mChunkServer.GetSessionKey();
            }
            // Roll forward access time to indicate the request is in flight.
            // If op fails or times out, then write restarts from write id
            // allocation.
            if (inOp.createChunkAccessFlag) {
                mChunkAccessExpireTime = theNow + LEASE_INTERVAL_SECS * 3 / 2;
            }
            if (inOp.createChunkServerAccessFlag) {
                mCSAccessExpireTime = theNow + LEASE_INTERVAL_SECS * 3 / 2;
            }
        }
        void Done(
            WriteIdAllocOp& inOp,
            bool            inCanceledFlag,
            IOBuffer*       inBufferPtr)
        {
            QCASSERT(&mWriteIdAllocOp == &inOp && ! inBufferPtr);
            mWriteIds.clear();
            if (inCanceledFlag) {
                return;
            }
            if (inOp.status < 0) {
                HandleError(inOp);
                return;
            }
            const size_t theServerCount = inOp.chunkServerLoc.size();
            mWriteIds.reserve(theServerCount);
            istringstream theStream(inOp.writeIdStr);
            for (size_t i = 0; i < theServerCount; i++) {
                WriteInfo theWInfo;
                if (! (theStream >>
                        theWInfo.serverLoc.hostname >>
                        theWInfo.serverLoc.port >>
                        theWInfo.writeId)) {
                    KFS_LOG_STREAM_ERROR << mLogPrefix <<
                        "write id alloc:"
                        " at index: "         << i <<
                        " of: "               << theServerCount <<
                        " invalid response: " << inOp.writeIdStr <<
                    KFS_LOG_EOM;
                    break;
                }
                mWriteIds.push_back(theWInfo);
            }
            if (theServerCount != mWriteIds.size()) {
                HandleError(inOp);
                return;
            }
            UpdateAccess(inOp);
            mLeaseExpireTime = Now() + LEASE_INTERVAL_SECS - 5;
            StartWrite();
        }
        void Write()
        {
            if (mOpDoneFlagPtr) {
                return;
            }
            bool theOpDoneFlag = false;
            mOpDoneFlagPtr = &theOpDoneFlag;
            Queue::Iterator theIt(mPendingQueue);
            WriteOp* theOpPtr;
            while (! mSleepingFlag &&
                    mErrorCode == 0 &&
                    mAllocOp.chunkId > 0 &&
                    (theOpPtr = theIt.Next())) {
                Write(*theOpPtr);
                if (theOpDoneFlag) {
                    return; // Unwind. "this" might be deleted.
                }
            }
            mOpDoneFlagPtr = 0;
        }
        void Write(
            WriteOp& inWriteOp)
        {
            while (inWriteOp.mBeginBlock < inWriteOp.mEndBlock) {
                if (mInFlightBlocks.test(inWriteOp.mBeginBlock)) {
                    return; // Wait until the in flight write done.
                }
                mInFlightBlocks.set(inWriteOp.mBeginBlock++, 1);
            }
            Reset(inWriteOp);
            inWriteOp.contentLength =
                size_t(inWriteOp.mBuffer.BytesConsumable());
            inWriteOp.mWritePrepareOp.chunkId            = mAllocOp.chunkId;
            inWriteOp.mWritePrepareOp.chunkVersion       =
                mAllocOp.chunkVersion;
            inWriteOp.mWritePrepareOp.writeInfo          = mWriteIds;
            inWriteOp.mWritePrepareOp.contentLength      =
                inWriteOp.contentLength;
            inWriteOp.mWritePrepareOp.numBytes           =
                inWriteOp.contentLength;
            inWriteOp.mWritePrepareOp.replyRequestedFlag =
                mWriteIdAllocOp.writePrepReplySupportedFlag;
            // No need to recompute checksums on retry. Presently the buffer
            // remains the unchanged.
            SetAccess(
                inWriteOp.mWritePrepareOp,
                inWriteOp.mWritePrepareOp.replyRequestedFlag
            );
            if (inWriteOp.mWritePrepareOp.replyRequestedFlag) {
                if (! inWriteOp.mChecksumValidFlag) {
                    inWriteOp.mWritePrepareOp.checksum = ComputeBlockChecksum(
                        &inWriteOp.mBuffer,
                        inWriteOp.mWritePrepareOp.numBytes
                    );<commit_msg>Kfs client library: fix write sync support with authentication on by copying / setting up chunk access with write sync op if / when write reply isn't supported. Write reply isn't supported only with old deprecated write protocol. With authentication on write reply should be always supported, unless it is turned off by configuration for testing purposes.<commit_after>//---------------------------------------------------------- -*- Mode: C++ -*-
// $Id$
//
// Created 2010/06/11
// Author: <NAME>
//
// Copyright 2010-2012 Quantcast Corp.
//
// This file is part of Kosmos File System (KFS).
//
// 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.
//
//
//----------------------------------------------------------------------------

#include "Writer.h"

#include <sstream>
#include <algorithm>
#include <cerrno>
#include <sstream>
#include <bitset>
#include <string.h>

#include "kfsio/IOBuffer.h"
#include "kfsio/NetManager.h"
#include "kfsio/checksum.h"
#include "kfsio/ITimeout.h"
#include "kfsio/ClientAuthContext.h"
#include "common/kfsdecls.h"
#include "common/MsgLogger.h"
#include "qcdio/QCUtils.h"
#include "qcdio/qcstutils.h"
#include "qcdio/qcdebug.h"
#include "qcdio/QCDLList.h"
#include "RSStriper.h"
#include "KfsOps.h"
#include "utils.h"
#include "KfsClient.h"

namespace KFS
{
namespace client
{

using std::min;
using std::max;
using std::string;
using std::ostream;
using std::ostringstream;
using std::istringstream;

// Kfs client write state machine implementation.
class Writer::Impl :
    public QCRefCountedObj,
    private ITimeout,
    private KfsNetClient::OpOwner
{
public:
    typedef QCRefCountedObj::StRef StRef;

    enum
    {
        kErrorNone       = 0,
        kErrorParameters = -EINVAL,
        kErrorTryAgain   = -EAGAIN,
        kErrorFault      = -EFAULT,
        kErrorNoEntry    = -ENOENT
    };

    Impl(
        Writer&     inOuter,
        MetaServer& inMetaServer,
        Completion* inCompletionPtr,
        int         inMaxRetryCount,
        int         inWriteThreshold,
        int         inMaxPartialBuffersCount,
        int         inTimeSecBetweenRetries,
        int         inOpTimeoutSec,
        int         inIdleTimeoutSec,
        int         inMaxWriteSize,
        string      inLogPrefix,
        int64_t     inChunkServerInitialSeqNum)
        : QCRefCountedObj(),
          ITimeout(),
          KfsNetClient::OpOwner(),
          mOuter(inOuter),
          mMetaServer(inMetaServer),
          mPathName(),
          mFileId(-1),
          mClosingFlag(false),
          mSleepingFlag(false),
          mErrorCode(0),
          mWriteThreshold(max(0, inWriteThreshold)),
          mPartialBuffersCount(0),
          mPendingCount(0),
          mIdleTimeoutSec(inIdleTimeoutSec),
          mOpTimeoutSec(inOpTimeoutSec),
          mMaxRetryCount(inMaxRetryCount),
          mTimeSecBetweenRetries(inTimeSecBetweenRetries),
          mMaxPartialBuffersCount(inMaxPartialBuffersCount),
          mMaxWriteSize(min((int)CHUNKSIZE,
            (int)((max(0, inMaxWriteSize) + CHECKSUM_BLOCKSIZE - 1) /
                CHECKSUM_BLOCKSIZE * CHECKSUM_BLOCKSIZE))),
          mReplicaCount(-1),
          mRetryCount(0),
          mFileSize(0),
          mOffset(0),
          mOpenChunkBlockSize(CHUNKSIZE),
          mChunkServerInitialSeqNum(inChunkServerInitialSeqNum),
          mCompletionPtr(inCompletionPtr),
          mBuffer(),
          mLogPrefix(inLogPrefix),
          mStats(),
          mNetManager(mMetaServer.GetNetManager()),
          mTruncateOp(0, 0, -1, 0),
          mOpStartTime(0),
          mCompletionDepthCount(0),
          mStriperProcessCount(0),
          mStriperPtr(0)
        { Writers::Init(mWriters); }
    int Open(
        kfsFileId_t inFileId,
        const char* inFileNamePtr,
        Offset      inFileSize,
        int         inStriperType,
        int         inStripeSize,
        int         inStripeCount,
        int         inRecoveryStripeCount,
        int         inReplicaCount)
    {
        if (inFileId <= 0 || ! inFileNamePtr || ! *inFileNamePtr) {
            return kErrorParameters;
        }
        if (mFileId > 0) {
            if (inFileId == mFileId &&
                    inFileNamePtr == mPathName) {
                return mErrorCode;
            }
            return kErrorParameters;
        }
        if (IsOpen() && mErrorCode != 0) {
            return (mErrorCode < 0 ? mErrorCode : -mErrorCode);
        }
        if (mClosingFlag || mSleepingFlag) {
            return kErrorTryAgain;
        }
        delete mStriperPtr;
        string theErrMsg;
        mStriperPtr = 0;
        mOpenChunkBlockSize = Offset(CHUNKSIZE);
        mStriperPtr = Striper::Create(
            inStriperType,
            inStripeCount,
            inRecoveryStripeCount,
            inStripeSize,
            inFileSize,
            mLogPrefix,
            *this,
            mOpenChunkBlockSize,
            theErrMsg
        );
        if (! theErrMsg.empty()) {
            KFS_LOG_STREAM_ERROR << mLogPrefix <<
                theErrMsg <<
            KFS_LOG_EOM;
            return kErrorParameters;
        }
        if (! mStriperPtr || mOpenChunkBlockSize < Offset(CHUNKSIZE)) {
            mOpenChunkBlockSize = Offset(CHUNKSIZE);
        }
        mBuffer.Clear();
        mStats.Clear();
        mReplicaCount          = inReplicaCount;
        mFileSize              = inFileSize;
        mPartialBuffersCount   = 0;
        mPathName              = inFileNamePtr;
        mErrorCode             = 0;
        mFileId                = inFileId;
        mTruncateOp.fid        = -1;
        mTruncateOp.pathname   = 0;
        mTruncateOp.fileOffset = mFileSize;
        mRetryCount            = 0;
        return StartWrite();
    }
    int Close()
    {
        if (! IsOpen()) {
            return 0;
        }
        if (mErrorCode != 0) {
            return mErrorCode;
        }
        if (mClosingFlag) {
            return kErrorTryAgain;
        }
        mClosingFlag = true;
        return StartWrite();
    }
    int Write(
        IOBuffer& inBuffer,
        int       inLength,
        Offset    inOffset,
        bool      inFlushFlag,
        int       inWriteThreshold)
    {
        if (inOffset < 0) {
            return kErrorParameters;
        }
        if (mErrorCode != 0) {
            return (mErrorCode < 0 ? mErrorCode : -mErrorCode);
        }
        if (mClosingFlag || ! IsOpen()) {
            return kErrorParameters;
        }
        if (inLength <= 0) {
            return (
                (ReportCompletion(0, inLength, inOffset) && inFlushFlag) ?
                StartWrite(true) : 0
            );
        }
        if (inOffset != mOffset + mBuffer.BytesConsumable()) {
            // Just flush for now, do not try to optimize buffer rewrite.
            const int thePrevRefCount = GetRefCount();
            const int theRet = Flush();
            if (theRet < 0) {
                return theRet;
            }
            if (thePrevRefCount > GetRefCount()) {
                return (mErrorCode < 0 ? mErrorCode : -mErrorCode);
            }
            mOffset = inOffset;
        }
        if (mMaxPartialBuffersCount == 0 ||
                inLength < IOBufferData::GetDefaultBufferSize() * 2) {
            // If write size is small, then copy it into the last buffer.
            mBuffer.ReplaceKeepBuffersFull(
                &inBuffer, mBuffer.BytesConsumable(), inLength);
        } else {
            if (mBuffer.IsEmpty()) {
                mPartialBuffersCount = 0;
            }
            mBuffer.Move(&inBuffer, inLength);
            mPartialBuffersCount++;
            if (mMaxPartialBuffersCount >= 0 &&
                    mPartialBuffersCount >= mMaxPartialBuffersCount) {
                mBuffer.MakeBuffersFull();
                mPartialBuffersCount = 0;
                mStats.mBufferCompactionCount++;
            }
        }
        if (inWriteThreshold >= 0) {
            mWriteThreshold = inWriteThreshold;
        }
        const int theErrorCode = StartWrite(inFlushFlag);
        return (theErrorCode == 0 ? inLength :
            (theErrorCode < 0 ? theErrorCode : -theErrorCode));
    }
    int Flush()
    {
        const int theErrorCode = StartWrite(true);
        return (theErrorCode < 0 ? theErrorCode : -theErrorCode);
    }
    void Stop()
    {
        while (! Writers::IsEmpty(mWriters)) {
            delete Writers::Front(mWriters);
        }
        if (mTruncateOp.fid >= 0) {
            mMetaServer.Cancel(&mTruncateOp, this);
        }
        if (mSleepingFlag) {
            mNetManager.UnRegisterTimeoutHandler(this);
            mSleepingFlag = false;
        }
        mClosingFlag = false;
        mBuffer.Clear();
    }
    void Shutdown()
    {
        Stop();
        mFileId      = -1;
        mErrorCode   = 0;
    }
    bool IsOpen() const
        { return (mFileId > 0); }
    bool IsClosing() const
        { return (IsOpen() && mClosingFlag); }
    bool IsActive() const
    {
        return (
            IsOpen() && (
                ! mBuffer.IsEmpty() ||
                ! Writers::IsEmpty(mWriters) ||
                mClosingFlag)
        );
    }
    Offset GetPendingSize() const
        { return (GetPendingSizeSelf() + mPendingCount); }
    int SetWriteThreshold(
        int inThreshold)
    {
        const int  theThreshold      = max(0, inThreshold);
        const bool theStartWriteFlag = mWriteThreshold > theThreshold;
        mWriteThreshold = theThreshold;
        return ((theStartWriteFlag && IsOpen() && mErrorCode == 0) ?
            StartWrite() : mErrorCode
        );
    }
    void DisableCompletion()
        { mCompletionPtr = 0; }
    void Register(
        Completion* inCompletionPtr)
    {
        if (inCompletionPtr == mCompletionPtr) {
            return;
        }
        if (mCompletionPtr) {
            mCompletionPtr->Unregistered(mOuter);
        }
        mCompletionPtr = inCompletionPtr;
    }
    bool Unregister(
        Completion* inCompletionPtr)
    {
        if (inCompletionPtr != mCompletionPtr) {
            return false;
        }
        mCompletionPtr = 0;
        return true;
    }
    void GetStats(
        Stats&               outStats,
        KfsNetClient::Stats& outChunkServersStats)
    {
        outStats             = mStats;
        outChunkServersStats = mChunkServersStats;
    }
    bool GetErrorCode() const
        { return mErrorCode; }

private:
    typedef KfsNetClient ChunkServer;

    class ChunkWriter : private ITimeout, private KfsNetClient::OpOwner
    {
    public:
        struct WriteOp;
        typedef QCDLList<WriteOp, 0> Queue;
        typedef QCDLList<ChunkWriter, 0> Writers;

        struct WriteOp : public KfsOp
        {
            WritePrepareOp mWritePrepareOp;
            WriteSyncOp    mWriteSyncOp;
            IOBuffer       mBuffer;
            size_t         mBeginBlock;
            size_t         mEndBlock;
            time_t         mOpStartTime;
            bool           mChecksumValidFlag;
            WriteOp*       mPrevPtr[1];
            WriteOp*       mNextPtr[1];

            WriteOp()
                : KfsOp(CMD_WRITE, 0),
                  mWritePrepareOp(0, 0, 0),
                  mWriteSyncOp(),
                  mBuffer(),
                  mBeginBlock(0),
                  mEndBlock(0),
                  mOpStartTime(0),
                  mChecksumValidFlag(false)
                { Queue::Init(*this); }
            void Delete(
                WriteOp** inListPtr)
            {
                Queue::Remove(inListPtr, *this);
                delete this;
            }
            virtual void Request(
                ostream& inStream)
            {
                if (mWritePrepareOp.replyRequestedFlag) {
                    mWritePrepareOp.seq = seq;
                } else {
                    mWriteSyncOp.seq    = seq;
                    mWritePrepareOp.seq = seq + 1;
                }
                mWritePrepareOp.Request(inStream);
            }
            virtual bool NextRequest(
                kfsSeq_t inSeqNum,
                ostream& inStream)
            {
                if (mWritePrepareOp.replyRequestedFlag) {
                    return false;
                }
                QCASSERT(seq <= inSeqNum && inSeqNum <= mWritePrepareOp.seq + 1);
                if (mWritePrepareOp.seq < inSeqNum) {
                    return false;
                }
                mWriteSyncOp.Request(inStream);
                return true;
            }
            virtual ostream& ShowSelf(
                ostream& inStream) const
            {
                inStream << mWritePrepareOp.Show();
                if (! mWritePrepareOp.replyRequestedFlag) {
                    inStream << " " << mWriteSyncOp.Show();
                }
                return inStream;
            }
            virtual void ParseResponseHeaderSelf(
                const Properties& inProps)
            {
                if (contentLength > 0) {
                    KFS_LOG_STREAM_ERROR <<
                        "invalid response content length: " << contentLength <<
                        " " << mWriteSyncOp.Show() <<
                    KFS_LOG_EOM;
                    contentLength = 0;
                }
                mWritePrepareOp.status    = status;
                mWritePrepareOp.statusMsg = statusMsg;
                mWriteSyncOp.status       = status;
                mWriteSyncOp.statusMsg    = statusMsg;
                if (mWritePrepareOp.replyRequestedFlag) {
                    mWritePrepareOp.ParseResponseHeaderSelf(inProps);
                } else {
                    mWriteSyncOp.ParseResponseHeaderSelf(inProps);
                }
            }
            void InitBlockRange()
            {
                QCASSERT(
                    mWritePrepareOp.offset >= 0 &&
                    mWritePrepareOp.offset +
                        mBuffer.BytesConsumable() <= (Offset)CHUNKSIZE
                );
                mBeginBlock = mWritePrepareOp.offset / CHECKSUM_BLOCKSIZE;
                mEndBlock   = mBeginBlock +
                    (mBuffer.BytesConsumable() + CHECKSUM_BLOCKSIZE - 1) /
                    CHECKSUM_BLOCKSIZE;
            }
        private:
            virtual ~WriteOp()
                {}
            WriteOp(
                const WriteOp& inWriteOp);
            WriteOp& operator=(
                const WriteOp& inWriteOp);
        };

        ChunkWriter(
            Impl&         inOuter,
            int64_t       inSeqNum,
            const string& inLogPrefix)
            : ITimeout(),
              KfsNetClient::OpOwner(),
              mOuter(inOuter),
              mChunkServer(
                inOuter.mNetManager,
                string(), -1,
                // All chunk server retries are handled here
                0, // inMaxRetryCount
                0, // inTimeSecBetweenRetries,
                inOuter.mOpTimeoutSec,
                inOuter.mIdleTimeoutSec,
                inSeqNum,
                inLogPrefix.c_str(),
                // Just fail the op. Error handler will reset connection and
                // cancel all pending ops by calling Stop()
                false // inResetConnectionOnOpTimeoutFlag
              ),
              mErrorCode(0),
              mRetryCount(0),
              mPendingCount(0),
              mOpenChunkBlockFileOffset(-1),
              mOpStartTime(0),
              mWriteIds(),
              mAllocOp(0, 0, ""),
              mWriteIdAllocOp(0, 0, 0, 0, 0),
              mCloseOp(0, 0),
              mLastOpPtr(0),
              mSleepingFlag(false),
              mClosingFlag(false),
              mLogPrefix(inLogPrefix),
              mOpDoneFlagPtr(0),
              mInFlightBlocks(),
              mHasSubjectIdFlag(),
              mChunkAccess(),
              mLeaseExpireTime(0),
              mChunkAccessExpireTime(0),
              mCSAccessExpireTime(0)
        {
            Queue::Init(mPendingQueue);
            Queue::Init(mInFlightQueue);
            Writers::Init(*this);
            Writers::PushFront(mOuter.mWriters, *this);
            mChunkServer.SetRetryConnectOnly(true);
            mAllocOp.fileOffset        = -1;
            mAllocOp.invalidateAllFlag = false;
        }
        ~ChunkWriter()
        {
            ChunkWriter::Shutdown();
            ChunkServer::Stats theStats;
            mChunkServer.GetStats(theStats);
            mOuter.mChunkServersStats.Add(theStats);
            Writers::Remove(mOuter.mWriters, *this);
        }
        void CancelClose()
            { mClosingFlag = false; }
        // The QueueWrite() guarantees that completion will not be invoked.
        // The writes will be queued even if the writer is already in the error
        // state: mErrorCode != 0. In the case of fatal error all pending writes
        // are discarded when the writer gets deleted.
        //
        // StartWrite() must be called in order to start executing pending
        // writes.
        // This allows the caller to properly update its state before the writes
        // get executed, and the corresponding completion(s) invoked.
        int QueueWrite(
            IOBuffer& inBuffer,
            int       inSize,
            Offset    inOffset,
            int       inWriteThreshold)
        {
            int theSize = min(inBuffer.BytesConsumable(), inSize);
            if (theSize <= 0) {
                return 0;
            }
            const Offset kChunkSize         = (Offset)CHUNKSIZE;
            const int    kChecksumBlockSize = (int)CHECKSUM_BLOCKSIZE;
            QCRTASSERT(inOffset >= 0 && ! mClosingFlag);
            const Offset theChunkOffset = inOffset % kChunkSize;
            if (mAllocOp.fileOffset < 0) {
                mAllocOp.fileOffset = inOffset - theChunkOffset;
                mOpenChunkBlockFileOffset = mAllocOp.fileOffset -
                    mAllocOp.fileOffset % mOuter.mOpenChunkBlockSize;
            } else {
                QCRTASSERT(mAllocOp.fileOffset == inOffset - theChunkOffset);
            }
            theSize = min(theSize, (int)(kChunkSize - theChunkOffset));
            QCASSERT(theSize > 0);
            Offset thePos = theChunkOffset;
            // Try to append to the last pending op.
            WriteOp* const theLastOpPtr = Queue::Back(mPendingQueue);
            if (theLastOpPtr) {
                WriteOp& theOp = *theLastOpPtr;
                const int    theOpSize = theOp.mBuffer.BytesConsumable();
                const Offset theOpPos  = theOp.mWritePrepareOp.offset;
                if (theOpPos + theOpSize == thePos) {
                    const int theHead = (int)(theOpPos % kChecksumBlockSize);
                    int       theNWr  = min(theSize,
                        (theHead == 0 ?
                            mOuter.mMaxWriteSize :
                            kChecksumBlockSize - theHead
                        ) - theOpSize
                    );
                    if (theNWr > 0 &&
                            theOpSize + theNWr > kChecksumBlockSize) {
                        theNWr -= (theOpSize + theNWr) % kChecksumBlockSize;
                    }
                    if (theNWr > 0) {
                        theOp.mBuffer.Move(&inBuffer, theNWr);
                        // Force checksum recomputation.
                        theOp.mChecksumValidFlag = false;
                        theOp.mWritePrepareOp.checksums.clear();
                        // Update last the block index.
                        const int theCurBegin = theOp.mBeginBlock;
                        theOp.InitBlockRange();
                        theOp.mBeginBlock = theCurBegin;
                        // The op is already in the pending queue.
                        theSize -= theNWr;
                        thePos  += theNWr;
                    }
                }
            }
            const int theWriteThreshold = thePos + theSize >= kChunkSize ?
                1 : max(inWriteThreshold, 1);
            const int theBlockOff       = (int)(thePos % kChecksumBlockSize);
            if (theBlockOff > 0 && (theSize >= theWriteThreshold ||
                    theBlockOff + theSize >= kChecksumBlockSize)) {
                WriteOp* const theWriteOpPtr = new WriteOp();
                theWriteOpPtr->mWritePrepareOp.offset = thePos;
                const int theNWr = theWriteOpPtr->mBuffer.Move(
                    &inBuffer,
                    min(theSize, kChecksumBlockSize - theBlockOff)
                );
                theSize -= theNWr;
                thePos  += theNWr;
                theWriteOpPtr->InitBlockRange();
                Queue::PushBack(mPendingQueue, *theWriteOpPtr);
            }
            while (theSize >= theWriteThreshold) {
                int theOpSize = min(mOuter.mMaxWriteSize, theSize);
                if (theOpSize > kChecksumBlockSize) {
                    theOpSize -= theOpSize % kChecksumBlockSize;
                }
                WriteOp* const theWriteOpPtr = new WriteOp();
                theWriteOpPtr->mWritePrepareOp.offset = thePos;
                const int theNWr =
                    theWriteOpPtr->mBuffer.Move(&inBuffer, theOpSize);
                theSize -= theNWr;
                thePos  += theNWr;
                theWriteOpPtr->InitBlockRange();
                Queue::PushBack(mPendingQueue, *theWriteOpPtr);
            }
            QCRTASSERT(thePos <= kChunkSize && theSize >= 0);
            const Offset theNWr = thePos - theChunkOffset;
            // The following must be updated before invoking StartWrite(),
            // as it could invoke completion immediately (in the case of
            // failure).
            mPendingCount += theNWr;
            return theNWr;
        }
        void StartWrite()
        {
            if (mSleepingFlag) {
                return;
            }
            if (mErrorCode != 0 && ! mAllocOp.invalidateAllFlag) {
                if (mLastOpPtr) {
                    Reset();
                }
                mClosingFlag = false;
                return;
            }
            if (mClosingFlag && ! CanWrite()) {
                if (! Queue::IsEmpty(mInFlightQueue)) {
                    return;
                }
                if (mLastOpPtr == &mCloseOp) {
                    return;
                }
                // Try to close chunk even if chunk server disconnected, to
                // release the write lease.
                if (mAllocOp.chunkId > 0) {
                    CloseChunk();
                    return;
                }
                mChunkServer.Stop();
                if (mLastOpPtr == &mAllocOp) {
                    mOuter.mMetaServer.Cancel(mLastOpPtr, this);
                }
                mClosingFlag        = false;
                mAllocOp.fileOffset = -1;
                mAllocOp.chunkId    = -1;
                ReportCompletion();
                return;
            }
            if (! CanWrite()) {
                return;
            }
            if (0 < mAllocOp.chunkId && mLeaseExpireTime < Now()) {
                // When chunk server disconnects it might clean up write lease.
                // Start from the beginning -- chunk allocation.
                KFS_LOG_STREAM_DEBUG << mLogPrefix <<
                    "write lease expired: " <<
                        mChunkServer.GetServerLocation() <<
                    " starting from chunk allocation, pending:" <<
                    " queue: " << (Queue::IsEmpty(mPendingQueue) ? "" : "not") <<
                        " empty" <<
                KFS_LOG_EOM;
                Reset();
                if (! CanWrite()) {
                    // Do not try to preallocate chunk after inactivity timeout
                    // or error, if no data pending.
                    return;
                }
            }
            // Return immediately after calling Write() and AllocateChunk(), as
            // these can invoke completion. Completion, in turn, can delete
            // this.
            // Other methods of this class have to return immediately (unwind)
            // after invoking StartWrite().
            if (mAllocOp.chunkId > 0 && ! mWriteIds.empty()) {
                Write();
            } else if (! mLastOpPtr) { // Close can be in flight.
                Reset();
                AllocateChunk();
            }
        }
        void Close()
        {
            if (! mClosingFlag && IsOpen()) {
                mClosingFlag = true;
                StartWrite();
            }
        }
        void Shutdown()
        {
            Reset();
            QCRTASSERT(Queue::IsEmpty(mInFlightQueue));
            while (! Queue::IsEmpty(mPendingQueue)) {
                Queue::Front(mPendingQueue)->Delete(mPendingQueue);
            }
            mClosingFlag  = false;
            mErrorCode    = 0;
            mPendingCount = 0;
        }
        Offset GetFileOffset() const
        {
            return (mErrorCode == 0 ? mAllocOp.fileOffset : -1);
        }
        bool IsIdle() const
        {
            return (
                Queue::IsEmpty(mPendingQueue) &&
                Queue::IsEmpty(mInFlightQueue) &&
                ! mClosingFlag
            );
        }
        bool IsOpen() const
        {
            return (
                mErrorCode == 0 &&
                mAllocOp.fileOffset >= 0 &&
                ! mClosingFlag
            );
        }
        int GetErrorCode() const
            { return mErrorCode; }
        Offset GetPendingCount() const
            { return mPendingCount; }
        ChunkWriter* GetPrevPtr()
        {
            ChunkWriter& thePrev = ChunkWritersListOp::GetPrev(*this);
            return (&thePrev == this ? 0 : &thePrev);
        }
        Offset GetOpenChunkBlockFileOffset() const
        {
            return (mAllocOp.fileOffset >= 0 ? mOpenChunkBlockFileOffset : -1);
        }

    private:
        typedef std::vector<WriteInfo> WriteIds;
        typedef std::bitset<CHUNKSIZE / CHECKSUM_BLOCKSIZE> ChecksumBlocks;

        Impl&          mOuter;
        ChunkServer    mChunkServer;
        int            mErrorCode;
        int            mRetryCount;
        Offset         mPendingCount;
        Offset         mOpenChunkBlockFileOffset;
        time_t         mOpStartTime;
        WriteIds       mWriteIds;
        AllocateOp     mAllocOp;
        WriteIdAllocOp mWriteIdAllocOp;
        CloseOp        mCloseOp;
        KfsOp*         mLastOpPtr;
        bool           mSleepingFlag;
        bool           mClosingFlag;
        string const   mLogPrefix;
        bool*          mOpDoneFlagPtr;
        ChecksumBlocks mInFlightBlocks;
        bool           mHasSubjectIdFlag;
        string         mChunkAccess;
        time_t         mLeaseExpireTime;
        time_t         mChunkAccessExpireTime;
        time_t         mCSAccessExpireTime;
        WriteOp*       mPendingQueue[1];
        WriteOp*       mInFlightQueue[1];
        ChunkWriter*   mPrevPtr[1];
        ChunkWriter*   mNextPtr[1];

        friend class QCDLListOp<ChunkWriter, 0>;
        typedef QCDLListOp<ChunkWriter, 0> ChunkWritersListOp;

        void AllocateChunk()
        {
            QCASSERT(
                mOuter.mFileId > 0 &&
                mAllocOp.fileOffset >= 0 &&
                ! Queue::IsEmpty(mPendingQueue)
            );
            Reset(mAllocOp);
            mAllocOp.fid                  = mOuter.mFileId;
            mAllocOp.pathname             = mOuter.mPathName;
            mAllocOp.append               = false;
            mAllocOp.chunkId              = -1;
            mAllocOp.chunkVersion         = -1;
            mAllocOp.spaceReservationSize = 0;
            mAllocOp.maxAppendersPerChunk = 0;
            mAllocOp.allowCSClearTextFlag = false;
            mAllocOp.chunkServerAccessValidForTime = 0;
            mAllocOp.chunkServerAccessIssuedTime   = 0;
            mAllocOp.chunkServers.clear();
            mAllocOp.chunkAccess.clear();
            mAllocOp.chunkServerAccessToken.clear();
            mOuter.mStats.mChunkAllocCount++;
            EnqueueMeta(mAllocOp);
        }
        void Done(
            AllocateOp& inOp,
            bool        inCanceledFlag,
            IOBuffer*   inBufferPtr)
        {
            QCASSERT(&mAllocOp == &inOp && ! inBufferPtr);
            if (inCanceledFlag) {
                return;
            }
            if (inOp.status != 0 || (mAllocOp.chunkServers.empty() &&
                    ! mAllocOp.invalidateAllFlag)) {
                mAllocOp.chunkId = 0;
                HandleError(inOp);
                return;
            }
            if (mAllocOp.invalidateAllFlag) {
                // Report all writes completed. Completion does not expect the
                // offset to match the original write offset with striper.
                KFS_LOG_STREAM_INFO << mLogPrefix <<
                    "invalidate done:"
                    " chunk: "   << mAllocOp.chunkId <<
                    " offset: "  << mAllocOp.fileOffset <<
                    " status: "  << inOp.status <<
                    " pending: " << mPendingCount <<
                    " w-empty: " << Queue::IsEmpty(mPendingQueue) <<
                KFS_LOG_EOM;
                const int    theSize   = mPendingCount;
                const Offset theOffset = theSize > 0 ? mAllocOp.fileOffset : 0;
                mAllocOp.invalidateAllFlag = false;
                Shutdown();
                ReportCompletion(theOffset, theSize);
                return;
            }
            mLeaseExpireTime = Now() + LEASE_INTERVAL_SECS - 5;
            AllocateWriteId();
        }
        bool CanWrite()
        {
            return (
                ! Queue::IsEmpty(mPendingQueue) ||
                mAllocOp.invalidateAllFlag
            );
        }
        void AllocateWriteId()
        {
            QCASSERT(mAllocOp.chunkId > 0 && ! mAllocOp.chunkServers.empty());
            Reset(mWriteIdAllocOp);
            mWriteIdAllocOp.chunkId                     = mAllocOp.chunkId;
            mWriteIdAllocOp.chunkVersion                = mAllocOp.chunkVersion;
            mWriteIdAllocOp.isForRecordAppend           = false;
            mWriteIdAllocOp.chunkServerLoc              = mAllocOp.chunkServers;
            mWriteIdAllocOp.offset                      = 0;
            mWriteIdAllocOp.numBytes                    = 0;
            mWriteIdAllocOp.writePrepReplySupportedFlag = false;

            const time_t theNow = Now();
            mHasSubjectIdFlag = false;
            mChunkAccess.clear();

            const bool theCSClearTextAllowedFlag =
                mOuter.IsChunkServerClearTextAllowed();
            mChunkServer.SetShutdownSsl(
                mAllocOp.allowCSClearTextFlag &&
                theCSClearTextAllowedFlag
            );
            if (mAllocOp.chunkServerAccessToken.empty() ||
                    mAllocOp.chunkAccess.empty()) {
                mChunkServer.SetKey(0, 0, 0, 0);
                mChunkServer.SetAuthContext(0);
                if (! mAllocOp.chunkServerAccessToken.empty()) {
                    mWriteIdAllocOp.status    = -EINVAL;
                    mWriteIdAllocOp.statusMsg = "no chunk access";
                } else if (! mAllocOp.chunkAccess.empty()) {
                    mWriteIdAllocOp.status    = -EINVAL;
                    mWriteIdAllocOp.statusMsg = "no chunk server access";
                } else if (! theCSClearTextAllowedFlag) {
                    mWriteIdAllocOp.status    = -EPERM;
                    mWriteIdAllocOp.statusMsg =
                        "no cleart text chunk server access";
                } else {
                    mChunkAccessExpireTime = theNow + 60 * 60 * 24 * 365;
                    mCSAccessExpireTime    = mChunkAccessExpireTime;
                }
            } else {
                mChunkServer.SetKey(
                    mAllocOp.chunkServerAccessToken.data(),
                    mAllocOp.chunkServerAccessToken.size(),
                    mAllocOp.chunkServerAccessKey.GetPtr(),
                    mAllocOp.chunkServerAccessKey.GetSize()
                );
                mChunkAccess           = mAllocOp.chunkAccess;
                mWriteIdAllocOp.access = mChunkAccess;
                // Always ask for chunk access token here, as the chunk access
                // token's lifetime returned by alloc is 5 min.
                // The chunk returns the token with the corresponding key's
                // lifetime as the token subject includes write id.
                mWriteIdAllocOp.createChunkAccessFlag = true;
                mChunkAccessExpireTime = theNow - 60 * 60 * 24;
                mCSAccessExpireTime = GetAccessExpireTime(
                    theNow,
                    mAllocOp.chunkServerAccessIssuedTime,
                    mAllocOp.chunkServerAccessValidForTime
                );
                mWriteIdAllocOp.createChunkServerAccessFlag =
                    mCSAccessExpireTime <= theNow;
                if (mAllocOp.allowCSClearTextFlag &&
                        theCSClearTextAllowedFlag &&
                        mWriteIdAllocOp.createChunkServerAccessFlag) {
                    mWriteIdAllocOp.decryptKey = &mChunkServer.GetSessionKey();
                }
                if (! mChunkServer.GetAuthContext()) {
                    mChunkServer.SetAuthContext(
                        mOuter.mMetaServer.GetAuthContext());
                }
            }
            if (mWriteIdAllocOp.status == 0) {
                const bool kCancelPendingOpsFlag = true;
                if (mChunkServer.SetServer(
                        mAllocOp.chunkServers[0],
                        kCancelPendingOpsFlag,
                        &mWriteIdAllocOp.statusMsg)) {
                    Enqueue(mWriteIdAllocOp);
                    return;
                }
                mWriteIdAllocOp.status = kErrorFault;
            }
            HandleError(mWriteIdAllocOp);
        }
        static int64_t GetAccessExpireTime(
            time_t  inNow,
            int64_t inIssedTime,
            int64_t inValidFor)
        {
            // Use current time if the clock difference is large enough.
            int64_t theDiff = inIssedTime - (int64_t)inNow;
            if (theDiff < 0) {
                theDiff = -theDiff;
            }
            return (
                ((LEASE_INTERVAL_SECS * 3 < theDiff) ? inNow : inIssedTime) +
                inValidFor - LEASE_INTERVAL_SECS
            );
        }
        void UpdateAccess(
            ChunkAccessOp& inOp)
        {
            if (! inOp.chunkAccessResponse.empty()) {
                mHasSubjectIdFlag      = true;
                mChunkAccess           = inOp.chunkAccessResponse;
                mChunkAccessExpireTime = GetAccessExpireTime(
                    Now(),
                    inOp.accessResponseIssued,
                    inOp.accessResponseValidForSec
                );
            }
            if (0 < inOp.accessResponseValidForSec &&
                    ! inOp.chunkServerAccessId.empty()) {
                mChunkServer.SetKey(
                    inOp.chunkServerAccessId.data(),
                    inOp.chunkServerAccessId.size(),
                    inOp.chunkServerAccessKey.GetPtr(),
                    inOp.chunkServerAccessKey.GetSize()
                );
                if (inOp.chunkAccessResponse.empty()) {
                    mCSAccessExpireTime = GetAccessExpireTime(
                        Now(),
                        inOp.accessResponseIssued,
                        inOp.accessResponseValidForSec
                    );
                } else {
                    mCSAccessExpireTime = mChunkAccessExpireTime;
                }
            }
        }
        void SetAccess(
            ChunkAccessOp& inOp,
            bool           inCanRequestAccessFlag = true)
        {
            const time_t theNow = Now();
            inOp.access                      = mChunkAccess;
            inOp.createChunkAccessFlag       = inCanRequestAccessFlag &&
                mChunkAccessExpireTime <= theNow;
            inOp.createChunkServerAccessFlag = inCanRequestAccessFlag &&
                mCSAccessExpireTime    <= theNow;
            inOp.hasSubjectIdFlag            =
                mHasSubjectIdFlag && ! mWriteIds.empty();
            if (inOp.hasSubjectIdFlag) {
                inOp.subjectId = mWriteIds.front().writeId;
            }
            if (inOp.createChunkServerAccessFlag &&
                    mChunkServer.IsShutdownSsl()) {
                inOp.decryptKey = &mChunkServer.GetSessionKey();
            }
            // Roll forward access time to indicate the request is in flight.
            // If op fails or times out, then write restarts from write id
            // allocation.
            if (inOp.createChunkAccessFlag) {
                mChunkAccessExpireTime = theNow + LEASE_INTERVAL_SECS * 3 / 2;
            }
            if (inOp.createChunkServerAccessFlag) {
                mCSAccessExpireTime = theNow + LEASE_INTERVAL_SECS * 3 / 2;
            }
        }
        void Done(
            WriteIdAllocOp& inOp,
            bool            inCanceledFlag,
            IOBuffer*       inBufferPtr)
        {
            QCASSERT(&mWriteIdAllocOp == &inOp && ! inBufferPtr);
            mWriteIds.clear();
            if (inCanceledFlag) {
                return;
            }
            if (inOp.status < 0) {
                HandleError(inOp);
                return;
            }
            const size_t theServerCount = inOp.chunkServerLoc.size();
            mWriteIds.reserve(theServerCount);
            istringstream theStream(inOp.writeIdStr);
            for (size_t i = 0; i < theServerCount; i++) {
                WriteInfo theWInfo;
                if (! (theStream >>
                        theWInfo.serverLoc.hostname >>
                        theWInfo.serverLoc.port >>
                        theWInfo.writeId)) {
                    KFS_LOG_STREAM_ERROR << mLogPrefix <<
                        "write id alloc:"
                        " at index: "         << i <<
                        " of: "               << theServerCount <<
                        " invalid response: " << inOp.writeIdStr <<
                    KFS_LOG_EOM;
                    break;
                }
                mWriteIds.push_back(theWInfo);
            }
            if (theServerCount != mWriteIds.size()) {
                HandleError(inOp);
                return;
            }
            UpdateAccess(inOp);
            mLeaseExpireTime = Now() + LEASE_INTERVAL_SECS - 5;
            StartWrite();
        }
        void Write()
        {
            if (mOpDoneFlagPtr) {
                return;
            }
            bool theOpDoneFlag = false;
            mOpDoneFlagPtr = &theOpDoneFlag;
            Queue::Iterator theIt(mPendingQueue);
            WriteOp* theOpPtr;
            while (! mSleepingFlag &&
                    mErrorCode == 0 &&
                    mAllocOp.chunkId > 0 &&
                    (theOpPtr = theIt.Next())) {
                Write(*theOpPtr);
                if (theOpDoneFlag) {
                    return; // Unwind. "this" might be deleted.
                }
            }
            mOpDoneFlagPtr = 0;
        }
        void Write(
            WriteOp& inWriteOp)
        {
            while (inWriteOp.mBeginBlock < inWriteOp.mEndBlock) {
                if (mInFlightBlocks.test(inWriteOp.mBeginBlock)) {
                    return; // Wait until the in flight write done.
                }
                mInFlightBlocks.set(inWriteOp.mBeginBlock++, 1);
            }
            Reset(inWriteOp);
            inWriteOp.contentLength =
                size_t(inWriteOp.mBuffer.BytesConsumable());
            inWriteOp.mWritePrepareOp.chunkId            = mAllocOp.chunkId;
            inWriteOp.mWritePrepareOp.chunkVersion       =
                mAllocOp.chunkVersion;
            inWriteOp.mWritePrepareOp.writeInfo          = mWriteIds;
            inWriteOp.mWritePrepareOp.contentLength      =
                inWriteOp.contentLength;
            inWriteOp.mWritePrepareOp.numBytes           =
                inWriteOp.contentLength;
            inWriteOp.mWritePrepareOp.replyRequestedFlag =
                mWriteIdAllocOp.writePrepReplySupportedFlag;
            // No need to recompute checksums on retry. Presently the buffer
            // remains the unchanged.
            SetAccess(
                inWriteOp.mWritePrepareOp,
                inWriteOp.mWritePrepareOp.replyRequestedFlag
            );
            if (! inWriteOp.mWritePrepareOp.replyRequestedFlag) {
                SetAccess(inWriteOp.mWriteSyncOp);
            }
            if (inWriteOp.mWritePrepareOp.replyRequestedFlag) {
                if (! inWriteOp.mChecksumValidFlag) {
                    inWriteOp.mWritePrepareOp.checksum = ComputeBlockChecksum(
                        &inWriteOp.mBuffer,
                        inWriteOp.mWritePrepareOp.numBytes
                    );
{'fraction_non_alphanumeric': 0.08476106770049484, 'fraction_numerical': 0.005249407033723829, 'mean_word_length': 1.3219367644181041, 'pattern_counts': {'":': 0, '<': 243, '<?xml version=': 0, '>': 119, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2614450', 'n_tokens_mistral': 22554, 'n_tokens_neox': 21516, 'n_words': 4351}