import org.apache.cassandra.utils.CLibrary; //导入依赖的package包/类
protected void syncInternal()
{
    if (syncNeeded)
    {
        flushInternal();
        syncDataOnlyInternal();

        if (!directorySynced)
        {
            CLibrary.trySync(directoryFD);
            directorySynced = true;
        }

        syncNeeded = false;
    }
}
 

import org.apache.cassandra.utils.CLibrary; //导入依赖的package包/类
@Override
public void close()
{
    if (buffer == null)
        return; // already closed

    syncInternal();

    buffer = null;

    if (skipIOCache && bytesSinceCacheFlush > 0)
        CLibrary.trySkipCache(fd, 0, 0);

    try
    {
        out.close();
    }
    catch (IOException e)
    {
        throw new FSWriteError(e, getPath());
    }

    FileUtils.closeQuietly(metadata);
    CLibrary.tryCloseFD(directoryFD);
}
 

import org.apache.cassandra.utils.CLibrary; //导入依赖的package包/类
@Override
void write(int startMarker, int nextMarker)
{
    // if there's room in the discard section to write an empty header,
    // zero out the next sync marker so replayer can cleanly exit
    if (nextMarker <= buffer.capacity() - SYNC_MARKER_SIZE)
    {
        buffer.putInt(nextMarker, 0);
        buffer.putInt(nextMarker + 4, 0);
    }

    // write previous sync marker to point to next sync marker
    // we don't chain the crcs here to ensure this method is idempotent if it fails
    writeSyncMarker(buffer, startMarker, startMarker, nextMarker);

    try {
        SyncUtil.force((MappedByteBuffer) buffer);
    }
    catch (Exception e) // MappedByteBuffer.force() does not declare IOException but can actually throw it
    {
        throw new FSWriteError(e, getPath());
    }
    CLibrary.trySkipCache(fd, startMarker, nextMarker, logFile.getAbsolutePath());
}
 

import org.apache.cassandra.utils.CLibrary; //导入依赖的package包/类
/**
 * Constructs a new segment file.
 *
 * @param filePath  if not null, recycles the existing file by renaming it and truncating it to CommitLog.SEGMENT_SIZE.
 */
CommitLogSegment(CommitLog commitLog)
{
    this.commitLog = commitLog;
    id = getNextId();
    descriptor = new CommitLogDescriptor(id, commitLog.configuration.getCompressorClass());
    logFile = new File(commitLog.location, descriptor.fileName());

    try
    {
        channel = FileChannel.open(logFile.toPath(), StandardOpenOption.WRITE, StandardOpenOption.READ, StandardOpenOption.CREATE);
        fd = CLibrary.getfd(channel);
    }
    catch (IOException e)
    {
        throw new FSWriteError(e, logFile);
    }

    buffer = createBuffer(commitLog);
    // write the header
    CommitLogDescriptor.writeHeader(buffer, descriptor);
    endOfBuffer = buffer.capacity();
    lastSyncedOffset = buffer.position();
    allocatePosition.set(lastSyncedOffset + SYNC_MARKER_SIZE);
}
 

import org.apache.cassandra.utils.CLibrary; //导入依赖的package包/类
@Override
public void close()
{
    if (buffer == null)
        return; // already closed

    syncInternal();

    buffer = null;

    try
    {
        out.close();
    }
    catch (IOException e)
    {
        throw new FSWriteError(e, getPath());
    }

    CLibrary.tryCloseFD(directoryFD);
}
 

import org.apache.cassandra.utils.CLibrary; //导入依赖的package包/类
public static void createHardLink(File from, File to)
{
    if (to.exists())
        throw new RuntimeException("Tried to create duplicate hard link to " + to);
    if (!from.exists())
        throw new RuntimeException("Tried to hard link to file that does not exist " + from);

    try
    {
        CLibrary.createHardLink(from, to);
    }
    catch (IOException e)
    {
        throw new FSWriteError(e, to);
    }
}
 

import org.apache.cassandra.utils.CLibrary; //导入依赖的package包/类
public void deallocate()
{
    buffer = null; // makes sure we don't use this after it's ostensibly closed

    if (skipIOCache && bytesSinceCacheFlush > 0)
        CLibrary.trySkipCache(fd, 0, 0);

    try
    {
        super.close();
    }
    catch (IOException e)
    {
        throw new FSReadError(e, filePath);
    }
}
 

import org.apache.cassandra.utils.CLibrary; //导入依赖的package包/类
@Override
public void close()
{
    if (buffer == null)
        return; // already closed

    syncInternal();

    buffer = null;

    if (skipIOCache && bytesSinceCacheFlush > 0)
        CLibrary.trySkipCache(fd, 0, 0);

    try
    {
        out.close();
    }
    catch (IOException e)
    {
        throw new FSWriteError(e, getPath());
    }

    CLibrary.tryCloseFD(directoryFD);
}
 

import org.apache.cassandra.utils.CLibrary; //导入依赖的package包/类
public void preheat(Map<DecoratedKey, RowIndexEntry> cachedKeys) throws IOException
{
    RandomAccessFile f = new RandomAccessFile(getFilename(), "r");

    try
    {
        int fd = CLibrary.getfd(f.getFD());

        for (Map.Entry<DecoratedKey, RowIndexEntry> entry : cachedKeys.entrySet())
        {
            cacheKey(entry.getKey(), entry.getValue());

            // add to the cache but don't do actual preheating if we have it disabled in the config
            if (DatabaseDescriptor.shouldPreheatPageCache() && fd > 0)
                CLibrary.preheatPage(fd, entry.getValue().position);
        }
    }
    finally
    {
        FileUtils.closeQuietly(f);
    }
}
 

import org.apache.cassandra.utils.CLibrary; //导入依赖的package包/类
public BufferedRandomAccessFile(File file, String mode, int bufferSize, boolean skipCache) throws IOException
{
    super(file, mode);

    this.skipCache = skipCache;

    channel = super.getChannel();
    filePath = file.getAbsolutePath();

    // allocating required size of the buffer
    if (bufferSize <= 0)
        throw new IllegalArgumentException("bufferSize must be positive");
    buffer = new byte[bufferSize];
    reBuffer();

    // if in read-only mode, caching file size
    fileLength = (mode.equals("r")) ? this.channel.size() : -1;
    fd = CLibrary.getfd(this.getFD());
}
 

import org.apache.cassandra.utils.CLibrary; //导入依赖的package包/类
/**
 * Flush (flush()) whatever writes are pending, and block until the data has been persistently committed (fsync()).
 */
public void sync() throws IOException
{
    if (syncNeeded)
    {
        flush();

        channel.force(true); // true, because file length counts as
                             // "meta-data"

        if (skipCache)
        {
            // clear entire file from page cache
            CLibrary.trySkipCache(this.fd, 0, 0);

            minBufferOffset = Long.MAX_VALUE;
            bytesSinceCacheFlush = 0;
        }

        syncNeeded = false;
    }
}
 

import org.apache.cassandra.utils.CLibrary; //导入依赖的package包/类
public void close()
{
    if (directoryDescriptor >= 0)
    {
        CLibrary.tryCloseFD(directoryDescriptor);
        directoryDescriptor = -1;
    }
}
 

import org.apache.cassandra.utils.CLibrary; //导入依赖的package包/类
public void tidy()
{
    lookup.remove(desc);
    if (readMeterSyncFuture != null)
        readMeterSyncFuture.cancel(true);
    if (isCompacted.get())
        SystemKeyspace.clearSSTableReadMeter(desc.ksname, desc.cfname, desc.generation);
    // don't ideally want to dropPageCache for the file until all instances have been released
    CLibrary.trySkipCache(desc.filenameFor(Component.DATA), 0, 0);
    CLibrary.trySkipCache(desc.filenameFor(Component.PRIMARY_INDEX), 0, 0);
}
 

import org.apache.cassandra.utils.CLibrary; //导入依赖的package包/类
public SSTableRewriter(ColumnFamilyStore cfs, Set<SSTableReader> rewriting, long maxAge, boolean isOffline)
{
    this.rewriting = rewriting;
    for (SSTableReader sstable : rewriting)
    {
        originalStarts.put(sstable.descriptor, sstable.first);
        fileDescriptors.put(sstable.descriptor, CLibrary.getfd(sstable.getFilename()));
    }
    this.dataTracker = cfs.getDataTracker();
    this.cfs = cfs;
    this.maxAge = maxAge;
    this.isOffline = isOffline;
}
 

import org.apache.cassandra.utils.CLibrary; //导入依赖的package包/类
protected void flushInternal()
{
    if (isDirty)
    {
        flushData();

        if (trickleFsync)
        {
            bytesSinceTrickleFsync += validBufferBytes;
            if (bytesSinceTrickleFsync >= trickleFsyncByteInterval)
            {
                syncDataOnlyInternal();
                bytesSinceTrickleFsync = 0;
            }
        }

        if (skipIOCache)
        {
            // we don't know when the data reaches disk since we aren't
            // calling flush
            // so we continue to clear pages we don't need from the first
            // offset we see
            // periodically we update this starting offset
            bytesSinceCacheFlush += validBufferBytes;

            if (bytesSinceCacheFlush >= RandomAccessReader.CACHE_FLUSH_INTERVAL_IN_BYTES)
            {
                CLibrary.trySkipCache(this.fd, ioCacheStartOffset, 0);
                ioCacheStartOffset = bufferOffset;
                bytesSinceCacheFlush = 0;
            }
        }

        // Remember that we wrote, so we don't write it again on next flush().
        resetBuffer();

        isDirty = false;
    }
}
 

import org.apache.cassandra.utils.CLibrary; //导入依赖的package包/类
void fsyncDirectory()
{
    int fd = CLibrary.tryOpenDirectory(hintsDirectory.getAbsolutePath());
    if (fd != -1)
    {
        SyncUtil.trySync(fd);
        CLibrary.tryCloseFD(fd);
    }
}
 

import org.apache.cassandra.utils.CLibrary; //导入依赖的package包/类
@SuppressWarnings("resource")
protected Page computeNext()
{
    CLibrary.trySkipCache(input.getChannel().getFileDescriptor(), 0, input.getFilePointer(), input.getPath());

    if (input.isEOF())
        return endOfData();

    return new Page(input.getFilePointer());
}
 

import org.apache.cassandra.utils.CLibrary; //导入依赖的package包/类
@SuppressWarnings("resource") // HintsWriter owns channel
static HintsWriter create(File directory, HintsDescriptor descriptor) throws IOException
{
    File file = new File(directory, descriptor.fileName());

    FileChannel channel = FileChannel.open(file.toPath(), StandardOpenOption.WRITE, StandardOpenOption.CREATE_NEW);
    int fd = CLibrary.getfd(channel);

    CRC32 crc = new CRC32();

    try (DataOutputBuffer dob = new DataOutputBuffer())
    {
        // write the descriptor
        descriptor.serialize(dob);
        ByteBuffer descriptorBytes = dob.buffer();
        updateChecksum(crc, descriptorBytes);
        channel.write(descriptorBytes);
    }
    catch (Throwable e)
    {
        channel.close();
        throw e;
    }

    if (descriptor.isCompressed())
    {
        return new CompressedHintsWriter(directory, descriptor, file, channel, fd, crc);
    }
    else
    {
        return new HintsWriter(directory, descriptor, file, channel, fd, crc);
    }
}
 

import org.apache.cassandra.utils.CLibrary; //导入依赖的package包/类
private void maybeSkipCache()
{
    long position = position();

    // don't skip page cache for tiny files, on the assumption that if they are tiny, the target node is probably
    // alive, and if so, the file will be closed and dispatched shortly (within a minute), and the file will be dropped.
    if (position >= DatabaseDescriptor.getTrickleFsyncIntervalInKb() * 1024L)
        CLibrary.trySkipCache(fd, 0, position - (position % PAGE_SIZE), file.getPath());
}
 

import org.apache.cassandra.utils.CLibrary; //导入依赖的package包/类
void close()
{
    if (folderDescriptor >= 0)
    {
        CLibrary.tryCloseFD(folderDescriptor);
        folderDescriptor = -1;
    }
}