chunkmanager.h

nandflash文件系统源代码

    /// Replay a chunk deletion
    /// @param[in] chunkId  Update the mChunkTable[] to remove this
    /// chunk id
    void ReplayDeleteChunk(kfsChunkId_t chunkId);


    /// Replay a write done on a chunk.
    /// @param[in] chunkId  Update the size of chunk to reflect the
    /// completion of a write.
    /// @param[in] chunkSize The new size of the chunk
    void ReplayWriteDone(kfsChunkId_t chunkId, off_t chunkSize,
                         off_t offset, std::vector<uint32_t> checksum);

    /// Replay a truncation done on a chunk.
    /// @param[in] chunkId  Update the size of chunk to reflect the
    /// completion of a truncation
    /// @param[in] chunkSize The new size of the chunk
    void ReplayTruncateDone(kfsChunkId_t chunkId, off_t chunkSize);

    /// Retrieve the chunks hosted on this chunk server.
    /// @param[out] result  A vector containing info of all chunks
    /// hosted on this server.
    void GetHostedChunks(std::vector<ChunkInfo_t> &result);

    /// Return the total space that is exported by this server.  If
    /// chunks are stored in a single directory, we use statvfs to
    /// determine the total space avail; we report the min of statvfs
    /// value and the configured mTotalSpace.
    int64_t GetTotalSpace();
    int64_t GetUsedSpace() const { return mUsedSpace; };
    long GetNumChunks() const { return mNumChunks; };

    /// For a write, the client is defining a write operation.  The op
    /// is queued and the client pushes data for it subsequently.
    /// @param[in] wi  The op that defines the write
    /// @retval status code
    int AllocateWriteId(WriteIdAllocOp *wi);

    /// For a write, the client has pushed data to us.  This is queued
    /// for a commit later on.
    /// @param[in] wp  The op that needs to be queued
    /// @retval status code
    int EnqueueWrite(WritePrepareOp *wp);

    /// Check if a write is pending to a chunk.
    /// @param[in] chunkId  The chunkid for which we are checking for
    /// pending write(s). 
    /// @retval True if a write is pending; false otherwise
    bool IsWritePending(kfsChunkId_t chunkId);

    /// Given a chunk id, return its version
    int64_t GetChunkVersion(kfsChunkId_t c);

    /// if the chunk exists and has a valid version #, then we need to page in the chunk meta-data.
    bool NeedToReadChunkMetadata(kfsChunkId_t c) {
        return GetChunkVersion(c) > 0;
    }

    /// Retrieve the write op given a write id.
    /// @param[in] writeId  The id corresponding to a previously
    /// enqueued write.
    /// @retval WriteOp if one exists; NULL otherwise
    WriteOp *GetWriteOp(int64_t writeId);

    /// The model with writes: allocate a write id (this causes a
    /// write-op to be created); then, push data for writes (which
    /// retrieves the write-op and then sends writes down to disk).
    /// The "clone" method makes a copy of a previously created
    /// write-op.
    /// @param[in] writeId the write id that was previously assigned
    /// @retval WriteOp if one exists; NULL otherwise
    WriteOp *CloneWriteOp(int64_t writeId);

    /// Set the status for a given write id
    void SetWriteStatus(int64_t writeId, int status);
    
    /// Is the write id a valid one
    bool IsValidWriteId(int64_t writeId);

    void Timeout();

    /// Push the changes from the write out to disk
    int Sync(WriteOp *op);

    /// return 0 if the chunkId is good; -EBADF otherwise
    int GetChunkChecksums(kfsChunkId_t chunkId, uint32_t **checksums) {
        ChunkInfoHandle_t *cih = NULL;
        
        if (GetChunkInfoHandle(chunkId, &cih) < 0)
            return -EBADF;
        *checksums = cih->chunkInfo.chunkBlockChecksum;
        return 0;
    }

    void ChunkIOFailed(kfsChunkId_t chunkId, int err) {
        if (err == -EIO) {
            NotifyMetaCorruptedChunk(chunkId);
            StaleChunk(chunkId);
        }
    }

private:
    /// How long should a pending write be held in LRU
    static const int MAX_PENDING_WRITE_LRU_SECS = 300;
    /// take a checkpoint once every 2 mins
    static const int CKPT_TIME_INTERVAL = 120;

    /// space available for allocation 
    int64_t	mTotalSpace;
    /// how much is used up by chunks
    int64_t	mUsedSpace;

    /// how many chunks are we hosting
    long	mNumChunks;

    time_t      mLastCheckpointTime;
    
    /// directories for storing the chunks
    std::vector<ChunkDirInfo_t> mChunkDirs;

    /// index of the last directory/drive that we used for placing a
    /// chunk
    int mLastDriveChosen;

    /// See the comments in KfsOps.cc near WritePreapreOp related to write handling
    int64_t mWriteId;
    std::list<WriteOp *> mPendingWrites;

    /// on a timeout, the timeout interface will force a checkpoint
    /// and query the disk manager for data
    ChunkManagerTimeoutImpl	*mChunkManagerTimeoutImpl;

    /// when taking checkpoints, write one out only if the chunk table
    /// is dirty. 
    bool mIsChunkTableDirty;
    /// table that maps chunkIds to their associated state
    CMap	mChunkTable;

    /// Given a chunk file name, extract out the
    /// fileid/chunkid/chunkversion from it and build a chunkinfo structure
    void MakeChunkInfoFromPathname(const std::string &pathname, off_t filesz, ChunkInfoHandle_t **result);

    /// Of the various directories this chunkserver is configured with, find the directory to store a chunk file.  
    /// This method does a "directory allocation".
    std::string GetDirForChunk();

    /// Utility function that given a chunkId, returns the full path
    /// to the chunk filename.
    std::string MakeChunkPathname(ChunkInfoHandle_t *cih);
    std::string MakeChunkPathname(const std::string &chunkdir, kfsFileId_t fid, kfsChunkId_t chunkId, kfsSeq_t chunkVersion);

    /// Utility function that given a chunkId, returns the full path
    /// to the chunk filename in the "stalechunks" dir
    std::string MakeStaleChunkPathname(ChunkInfoHandle_t *cih);

    /// update the used space in the directory where the chunk resides by nbytes.
    void UpdateDirSpace(ChunkInfoHandle_t *cih, off_t nbytes);

    /// Utility function that sets up a disk connection for an
    /// I/O operation on a chunk.
    /// @param[in] chunkId  Id of the chunk on which we are doing I/O
    /// @param[in] op   The KfsOp that is being on the chunk
    /// @retval A disk connection pointer allocated via a call to new;
    /// it is the caller's responsibility to free the memory
    DiskConnection *SetupDiskConnection(kfsChunkId_t chunkId, KfsOp *op);

    /// Utility function that returns a pointer to mChunkTable[chunkId].
    /// @param[in] chunkId  the chunk id for which we want info
    /// @param[out] cih  the resulting pointer from mChunkTable[chunkId]
    /// @retval  0 on success; -EBADF if we can't find mChunkTable[chunkId]
    int GetChunkInfoHandle(kfsChunkId_t chunkId, ChunkInfoHandle_t **cih);

    /// Checksums are computed on 64K blocks.  To verify checksums on
    /// reads, reads are aligned at 64K boundaries and data is read in
    /// 64K blocks.  So, for reads that are un-aligned/read less data,
    /// adjust appropriately.
    void AdjustDataRead(ReadOp *op);

    /// Pad the buffer with sufficient 0's so that checksumming works
    /// out.
    /// @param[in/out] buffer  The buffer to be padded with 0's
    void ZeroPad(IOBuffer *buffer);

    /// Given a chunkId and offset, return the checksum of corresponding
    /// "checksum block"---i.e., the 64K block that contains offset.
    uint32_t GetChecksum(kfsChunkId_t chunkId, off_t offset);

    /// For any writes that have been held for more than 2 mins,
    /// scavenge them and reclaim memory.
    void ScavengePendingWrites();

    /// If we have too many open fd's close out whatever we can.  When
    /// periodic is set, we do a scan and clean up.
    void CleanupInactiveFds(bool periodic = false);

    /// Notify the metaserver that chunk chunkId is corrupted; the
    /// metaserver will re-replicate this chunk and for now, won't
    /// send us traffic for this chunk.
    void NotifyMetaCorruptedChunk(kfsChunkId_t chunkId);

    /// For some reason, dirname is not accessable (for instance, the
    /// drive may have failed); in this case, notify metaserver that
    /// all the blocks on that dir are lost and the metaserver can
    /// then re-replicate.
    void NotifyMetaChunksLost(const std::string &dirname);

    /// Get all the chunk filenames into a single array.
    /// @retval on success, # of entries in the array;
    ///         on failures, -1
    int GetChunkDirsEntries(struct dirent ***namelist);
    /// Get all the chunk pathnames into a single vector
    void GetChunkPathEntries(std::vector<std::string> &pathnames);

    /// Helper function to move a chunk to the stale dir
    void MarkChunkStale(ChunkInfoHandle_t *cih);


    /// Scan the chunk dirs and rebuild the list of chunks that are hosted on this server
    void Restore();
    /// Restore the chunk meta-data from the specified file name.
    void RestoreChunkMeta(const std::string &chunkMetaFn);
    
    /// Update the checksums in the chunk metadata based on the op.
    void UpdateChecksums(ChunkInfoHandle_t *cih, WriteOp *op);
};

/// A Timeout interface object for taking checkpoints on the
/// ChunkManager object.
class ChunkManagerTimeoutImpl : public ITimeout {
public:
    ChunkManagerTimeoutImpl(ChunkManager *mgr) : mTimeoutOp(0) {
        mChunkManager = mgr; 
        // set a checkpoint once every min.
        // SetTimeoutInterval(60*1000);
    };
    void Timeout();
private:
    /// Owning chunk manager
    ChunkManager	*mChunkManager;
    TimeoutOp		mTimeoutOp;
};

extern ChunkManager gChunkManager;

/// Given a partition that holds chunks, get the path to the directory
/// that is used to keep the stale chunks (from this partition)
std::string GetStaleChunkPath(const std::string &partition);

}

#endif // _CHUNKMANAGER_H