flume-issues mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From 翟玉勇 (JIRA) <j...@apache.org>
Subject [jira] [Updated] (FLUME-3266) the parameters of fsyncPerTransaction are not working
Date Thu, 09 Aug 2018 09:41:00 GMT

     [ https://issues.apache.org/jira/browse/FLUME-3266?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]

翟玉勇 updated FLUME-3266:
-----------------------
    Description: 
every take or put transaction has sync disk?
{code}
this.fsyncPerTransaction = fsyncPerTransaction;
      this.fsyncInterval = fsyncInterval;
      if (!fsyncPerTransaction) {
        LOG.info("Sync interval = " + fsyncInterval);
        syncExecutor = Executors.newSingleThreadScheduledExecutor();
        syncExecutor.scheduleWithFixedDelay(new Runnable() {
          @Override
          public void run() {
            try {
              sync("Thread-other");
            } catch (Throwable ex) {
              LOG.error("Data file, " + getFile().toString() + " could not " +
                  "be synced to disk due to an error.", ex);
            }
          }
        }, fsyncInterval, fsyncInterval, TimeUnit.SECONDS);
      } else {
        syncExecutor = null;
      }
{code}


{code}
private void commit(long transactionID, short type) throws IOException {
    Preconditions.checkState(open, "Log is closed");
    Commit commit = new Commit(transactionID, WriteOrderOracle.next(), type);
    ByteBuffer buffer = TransactionEventRecord.toByteBuffer(commit);
    int logFileIndex = nextLogWriter(transactionID);
    long usableSpace = logFiles.get(logFileIndex).getUsableSpace();
    long requiredSpace = minimumRequiredSpace + buffer.limit();
    if (usableSpace <= requiredSpace) {
      throw new IOException("Usable space exhausted, only " + usableSpace +
          " bytes remaining, required " + requiredSpace + " bytes");
    }
    boolean error = true;
    try {
      try {
        LogFile.Writer logFileWriter = logFiles.get(logFileIndex);
        // If multiple transactions are committing at the same time,
        // this ensures that the number of actual fsyncs is small and a
        // number of them are grouped together into one.
        logFileWriter.commit(buffer);
        logFileWriter.sync();
        error = false;
      } catch (LogFileRetryableIOException e) {
        if (!open) {
          throw e;
        }
        roll(logFileIndex, buffer);
        LogFile.Writer logFileWriter = logFiles.get(logFileIndex);
        logFileWriter.commit(buffer);
        logFileWriter.sync();
        error = false;
      }
    } finally {
      if (error && open) {
        roll(logFileIndex);
      }
    }
  }
{code}



  was:
every take or put transaction has sync disk?
{code}
this.fsyncPerTransaction = fsyncPerTransaction;
      this.fsyncInterval = fsyncInterval;
      if (!fsyncPerTransaction) {
        LOG.info("Sync interval = " + fsyncInterval);
        syncExecutor = Executors.newSingleThreadScheduledExecutor();
        syncExecutor.scheduleWithFixedDelay(new Runnable() {
          @Override
          public void run() {
            try {
              sync("Thread-other");
            } catch (Throwable ex) {
              LOG.error("Data file, " + getFile().toString() + " could not " +
                  "be synced to disk due to an error.", ex);
            }
          }
        }, fsyncInterval, fsyncInterval, TimeUnit.SECONDS);
      } else {
        syncExecutor = null;
      }
{code}
private void commit(long transactionID, short type) throws IOException {
    Preconditions.checkState(open, "Log is closed");
    Commit commit = new Commit(transactionID, WriteOrderOracle.next(), type);
    ByteBuffer buffer = TransactionEventRecord.toByteBuffer(commit);
    int logFileIndex = nextLogWriter(transactionID);
    long usableSpace = logFiles.get(logFileIndex).getUsableSpace();
    long requiredSpace = minimumRequiredSpace + buffer.limit();
    if (usableSpace <= requiredSpace) {
      throw new IOException("Usable space exhausted, only " + usableSpace +
          " bytes remaining, required " + requiredSpace + " bytes");
    }
    boolean error = true;
    try {
      try {
        LogFile.Writer logFileWriter = logFiles.get(logFileIndex);
        // If multiple transactions are committing at the same time,
        // this ensures that the number of actual fsyncs is small and a
        // number of them are grouped together into one.
        logFileWriter.commit(buffer);
        logFileWriter.sync();
        error = false;
      } catch (LogFileRetryableIOException e) {
        if (!open) {
          throw e;
        }
        roll(logFileIndex, buffer);
        LogFile.Writer logFileWriter = logFiles.get(logFileIndex);
        logFileWriter.commit(buffer);
        logFileWriter.sync();
        error = false;
      }
    } finally {
      if (error && open) {
        roll(logFileIndex);
      }
    }
  }





> the parameters of fsyncPerTransaction are not working 
> ------------------------------------------------------
>
>                 Key: FLUME-3266
>                 URL: https://issues.apache.org/jira/browse/FLUME-3266
>             Project: Flume
>          Issue Type: Bug
>          Components: File Channel
>    Affects Versions: 1.8.0
>            Reporter: 翟玉勇
>            Priority: Minor
>
> every take or put transaction has sync disk?
> {code}
> this.fsyncPerTransaction = fsyncPerTransaction;
>       this.fsyncInterval = fsyncInterval;
>       if (!fsyncPerTransaction) {
>         LOG.info("Sync interval = " + fsyncInterval);
>         syncExecutor = Executors.newSingleThreadScheduledExecutor();
>         syncExecutor.scheduleWithFixedDelay(new Runnable() {
>           @Override
>           public void run() {
>             try {
>               sync("Thread-other");
>             } catch (Throwable ex) {
>               LOG.error("Data file, " + getFile().toString() + " could not " +
>                   "be synced to disk due to an error.", ex);
>             }
>           }
>         }, fsyncInterval, fsyncInterval, TimeUnit.SECONDS);
>       } else {
>         syncExecutor = null;
>       }
> {code}
> {code}
> private void commit(long transactionID, short type) throws IOException {
>     Preconditions.checkState(open, "Log is closed");
>     Commit commit = new Commit(transactionID, WriteOrderOracle.next(), type);
>     ByteBuffer buffer = TransactionEventRecord.toByteBuffer(commit);
>     int logFileIndex = nextLogWriter(transactionID);
>     long usableSpace = logFiles.get(logFileIndex).getUsableSpace();
>     long requiredSpace = minimumRequiredSpace + buffer.limit();
>     if (usableSpace <= requiredSpace) {
>       throw new IOException("Usable space exhausted, only " + usableSpace +
>           " bytes remaining, required " + requiredSpace + " bytes");
>     }
>     boolean error = true;
>     try {
>       try {
>         LogFile.Writer logFileWriter = logFiles.get(logFileIndex);
>         // If multiple transactions are committing at the same time,
>         // this ensures that the number of actual fsyncs is small and a
>         // number of them are grouped together into one.
>         logFileWriter.commit(buffer);
>         logFileWriter.sync();
>         error = false;
>       } catch (LogFileRetryableIOException e) {
>         if (!open) {
>           throw e;
>         }
>         roll(logFileIndex, buffer);
>         LogFile.Writer logFileWriter = logFiles.get(logFileIndex);
>         logFileWriter.commit(buffer);
>         logFileWriter.sync();
>         error = false;
>       }
>     } finally {
>       if (error && open) {
>         roll(logFileIndex);
>       }
>     }
>   }
> {code}



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@flume.apache.org
For additional commands, e-mail: issues-help@flume.apache.org


Mime
View raw message