fixbug/install snapshot bug

jraft-core/src/main/java/com/alipay/sofa/jraft/option/RpcOptions.java

@@ -32,6 +32,12 @@ public class RpcOptions {
  */
  private int rpcDefaultTimeout = 5000;
 
+ /**
+ * Install snapshot RPC request default timeout in milliseconds
+ * Default: 5 * 60 * 1000(5min)
+ */
+ private int rpcInstallSnapshotTimeout = 5 * 60 * 1000;
+
  /**
  * Rpc process thread pool size
  * Default: 80
@@ -75,10 +81,19 @@ public void setRpcDefaultTimeout(int rpcDefaultTimeout) {
  this.rpcDefaultTimeout = rpcDefaultTimeout;
  }
 
+ public int getRpcInstallSnapshotTimeout() {
+ return rpcInstallSnapshotTimeout;
+ }
+
+ public void setRpcInstallSnapshotTimeout(int rpcInstallSnapshotTimeout) {
+ this.rpcInstallSnapshotTimeout = rpcInstallSnapshotTimeout;
+ }
+
  @Override
  public String toString() {
  return "RpcOptions{" + "rpcConnectTimeoutMs=" + rpcConnectTimeoutMs + ", rpcDefaultTimeout="
- + rpcDefaultTimeout + ", rpcProcessorThreadPoolSize=" + rpcProcessorThreadPoolSize + ", metricRegistry="
- + metricRegistry + '}';
+ + rpcDefaultTimeout + ", rpcInstallSnapshotTimeout=" + rpcInstallSnapshotTimeout
+ + ", rpcProcessorThreadPoolSize=" + rpcProcessorThreadPoolSize + ", metricRegistry=" + metricRegistry
+ + '}';
  }
 }

jraft-core/src/main/java/com/alipay/sofa/jraft/rpc/impl/core/BoltRaftClientService.java

@@ -100,7 +100,7 @@ public Future<Message> getFile(Endpoint endpoint, GetFileRequest request, int ti
  @Override
  public Future<Message> installSnapshot(Endpoint endpoint, InstallSnapshotRequest request,
  RpcResponseClosure<InstallSnapshotResponse> done) {
- return invokeWithDone(endpoint, request, done, rpcOptions.getRpcDefaultTimeout());
+ return invokeWithDone(endpoint, request, done, rpcOptions.getRpcInstallSnapshotTimeout());
  }
 
  @Override

jraft-core/src/main/java/com/alipay/sofa/jraft/storage/FileService.java

@@ -55,7 +55,7 @@ public final class FileService {
  private static final FileService INSTANCE = new FileService();
 
  private final ConcurrentMap<Long, FileReader> fileReaderMap = new ConcurrentHashMap<>();
- private final AtomicLong nextId;
+ private final AtomicLong nextId = new AtomicLong();
 
  /**
  * Retrieve the singleton instance of FileService.
@@ -72,18 +72,16 @@ void clear() {
  }
 
  private FileService() {
- this.nextId = new AtomicLong();
- final long initValue = Math.abs(Utils.getProcessId(ThreadLocalRandom.current().nextLong(10000,
- Integer.MAX_VALUE)) << 45
- | System.nanoTime() << 17 >> 17);
- this.nextId.set(initValue);
- LOG.info("Initial file reader id in FileService is {}", this.nextId);
+ final long processId = Utils.getProcessId(ThreadLocalRandom.current().nextLong(10000, Integer.MAX_VALUE));
+ final long initialValue = Math.abs(processId << 45 | System.nanoTime() << 17 >> 17);
+ this.nextId.set(initialValue);
+ LOG.info("Initial file reader id in FileService is {}", initialValue);
  }
 
  /**
- * Handle GetFileRequest ,run the response or set the response with done.
+ * Handle GetFileRequest, run the response or set the response with done.
  */
- public Message handleGetFile(GetFileRequest request, RpcRequestClosure done) {
+ public Message handleGetFile(final GetFileRequest request, final RpcRequestClosure done) {
  if (request.getCount() <= 0 || request.getOffset() < 0) {
  return RpcResponseFactory.newResponse(RaftError.EREQUEST, "Invalid request: %s", request);
  }
@@ -92,23 +90,24 @@ public Message handleGetFile(GetFileRequest request, RpcRequestClosure done) {
  return RpcResponseFactory.newResponse(RaftError.ENOENT, "Fail to find reader=%d", request.getReaderId());
  }
 
- LOG.debug("GetFile from {} path={} filename={} offset={} count={}", done.getBizContext().getRemoteAddress(),
- reader.getPath(), request.getFilename(), request.getOffset(), request.getCount());
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("GetFile from {} path={} filename={} offset={} count={}",
+ done.getBizContext().getRemoteAddress(), reader.getPath(), request.getFilename(), request.getOffset(),
+ request.getCount());
+ }
 
  final ByteBufferCollector dataBuffer = ByteBufferCollector.allocate();
  final GetFileResponse.Builder responseBuilder = GetFileResponse.newBuilder();
  try {
  final int read = reader
  .readFile(dataBuffer, request.getFilename(), request.getOffset(), request.getCount());
  responseBuilder.setReadSize(read);
- if (read == -1) {
- responseBuilder.setEof(true);
- }
+ responseBuilder.setEof(read == FileReader.EOF);
  final ByteBuffer buf = dataBuffer.getBuffer();
  buf.flip();
  if (!buf.hasRemaining()) {
  // skip empty data
- return responseBuilder.setData(ByteString.EMPTY).build();
+ responseBuilder.setData(ByteString.EMPTY);
  } else {
  // TODO check hole
  responseBuilder.setData(ZeroByteStringHelper.wrap(buf));
@@ -128,9 +127,9 @@ public Message handleGetFile(GetFileRequest request, RpcRequestClosure done) {
  /**
  * Adds a file reader and return it's generated readerId.
  */
- public long addReader(FileReader reader) {
+ public long addReader(final FileReader reader) {
  final long readerId = this.nextId.getAndIncrement();
- if (fileReaderMap.putIfAbsent(readerId, reader) == null) {
+ if (this.fileReaderMap.putIfAbsent(readerId, reader) == null) {
  return readerId;
  } else {
  return -1L;
@@ -140,7 +139,7 @@ public long addReader(FileReader reader) {
  /**
  * Remove the reader by readerId.
  */
- public boolean removeReader(long readerId) {
+ public boolean removeReader(final long readerId) {
  return this.fileReaderMap.remove(readerId) != null;
  }
 }

jraft-core/src/main/java/com/alipay/sofa/jraft/storage/io/FileReader.java

@@ -30,6 +30,8 @@
  */
 public interface FileReader {
 
+ int EOF = -1;
+
  /**
  * Get the file path.
  *
@@ -39,8 +41,17 @@ public interface FileReader {
 
  /**
  * Read file into buf starts from offset at most maxCount.
- * Returns -1 if reaches end, else return read count.
+ *
+ * @param buf read bytes into this buf
+ * @param fileName file name
+ * @param offset the offset of file
+ * @param maxCount max read bytes
+ * @return -1 if reaches end, else return read count.
+ * @throws IOException if some I/O error occurs
+ * @throws RetryAgainException if it's not allowed to read partly
+ * or it's allowed but throughput is throttled to 0, try again.
  */
- int readFile(ByteBufferCollector buf, String fileName, long offset, long maxCount) throws IOException,
- RetryAgainException;
+ int readFile(final ByteBufferCollector buf, final String fileName, final long offset, final long maxCount)
+ throws IOException,
+ RetryAgainException;
 }

jraft-core/src/main/java/com/alipay/sofa/jraft/storage/io/LocalDirReader.java

@@ -29,12 +29,14 @@
 import com.google.protobuf.Message;
 
 /**
- * read a file data form local dir by fileName.
+ * Read a file data form local dir by fileName.
+ *
  * @author boyan (boyan@alibaba-inc.com)
  *
  * 2018-Apr-06 9:25:12 PM
  */
 public class LocalDirReader implements FileReader {
+
  private static final Logger LOG = LoggerFactory.getLogger(LocalDirReader.class);
 
  private final String path;
@@ -46,44 +48,45 @@ public LocalDirReader(String path) {
 
  @Override
  public String getPath() {
- return this.path;
+ return path;
  }
 
  @Override
- public int readFile(ByteBufferCollector buf, String fileName, long offset, long maxCount) throws IOException,
- RetryAgainException {
- return this.readFileWithMeta(buf, fileName, null, offset, maxCount);
+ public int readFile(final ByteBufferCollector buf, final String fileName, final long offset, final long maxCount)
+ throws IOException,
+ RetryAgainException {
+ return readFileWithMeta(buf, fileName, null, offset, maxCount);
  }
 
  @SuppressWarnings("unused")
- protected int readFileWithMeta(ByteBufferCollector buf, String fileName, Message fileMeta, long offset,
- long maxCount) throws IOException, RetryAgainException {
+ protected int readFileWithMeta(final ByteBufferCollector buf, final String fileName, final Message fileMeta,
+ long offset, final long maxCount) throws IOException, RetryAgainException {
  buf.expandIfNecessary();
  final String filePath = this.path + File.separator + fileName;
  final File file = new File(filePath);
- try (FileInputStream input = new FileInputStream(file); FileChannel fc = input.getChannel()) {
+ try (final FileInputStream input = new FileInputStream(file); final FileChannel fc = input.getChannel()) {
  int totalRead = 0;
  while (true) {
  final int nread = fc.read(buf.getBuffer(), offset);
  if (nread <= 0) {
- return -1;
+ return EOF;
  }
  totalRead += nread;
  if (totalRead < maxCount) {
  if (buf.hasRemaining()) {
- return -1;
+ return EOF;
  } else {
  buf.expandAtMost((int) (maxCount - totalRead));
  offset += nread;
  }
  } else {
  final long fsize = file.length();
  if (fsize < 0) {
- LOG.warn("Invlaid file length {}", filePath);
- return -1;
+ LOG.warn("Invalid file length {}", filePath);
+ return EOF;
  }
- if (fsize == offset + maxCount) {
- return -1;
+ if (fsize == offset + nread) {
+ return EOF;
  } else {
  return totalRead;
  }

jraft-core/src/main/java/com/alipay/sofa/jraft/storage/snapshot/local/SnapshotFileReader.java

@@ -54,21 +54,20 @@ public SnapshotFileReader(String path, SnapshotThrottle snapshotThrottle) {
  }
 
  public boolean open() {
- final File file = new File(this.getPath());
+ final File file = new File(getPath());
  return file.exists();
  }
 
  @Override
- public int readFile(ByteBufferCollector metaBufferCollector, String fileName, long offset, long maxCount)
- throws IOException,
- RetryAgainException {
+ public int readFile(final ByteBufferCollector metaBufferCollector, final String fileName, final long offset,
+ final long maxCount) throws IOException, RetryAgainException {
  // read the whole meta file.
  if (fileName.equals(Snapshot.JRAFT_SNAPSHOT_META_FILE)) {
  final ByteBuffer metaBuf = this.metaTable.saveToByteBufferAsRemote();
- //because bufRef will flip the buffer before using, so we must set the meta buffer position to it's limit.
+ // because bufRef will flip the buffer before using, so we must set the meta buffer position to it's limit.
  metaBuf.position(metaBuf.limit());
  metaBufferCollector.setBuffer(metaBuf);
- return -1;
+ return EOF;
  }
  final LocalFileMeta fileMeta = this.metaTable.getFileMeta(fileName);
  if (fileMeta == null) {
@@ -78,7 +77,7 @@ public int readFile(ByteBufferCollector metaBufferCollector, String fileName, lo
  // go through throttle
  long newMaxCount = maxCount;
  if (this.snapshotThrottle != null) {
- newMaxCount = snapshotThrottle.throttledByThroughput(maxCount);
+ newMaxCount = this.snapshotThrottle.throttledByThroughput(maxCount);
  if (newMaxCount < maxCount) {
  // if it's not allowed to read partly or it's allowed but
  // throughput is throttled to 0, try again.
@@ -88,7 +87,7 @@ public int readFile(ByteBufferCollector metaBufferCollector, String fileName, lo
  }
  }
 
- return this.readFileWithMeta(metaBufferCollector, fileName, fileMeta, offset, newMaxCount);
+ return super.readFileWithMeta(metaBufferCollector, fileName, fileMeta, offset, newMaxCount);
  }
 
 }