diff --git a/HISTORY b/HISTORY index aff4406..62e6193 100644 --- a/HISTORY +++ b/HISTORY @@ -1,5 +1,5 @@ -Version 1.38 2025-11-27 +Version 1.38 2025-11-28 * bugfixed: loadStorageServersFromTracker correctly with rw option * throw IOException with msg "connection broken" when in.read() < 0 diff --git a/src/main/java/org/csource/fastdfs/ProtoCommon.java b/src/main/java/org/csource/fastdfs/ProtoCommon.java index 273d7fb..4d8dd16 100644 --- a/src/main/java/org/csource/fastdfs/ProtoCommon.java +++ b/src/main/java/org/csource/fastdfs/ProtoCommon.java @@ -15,6 +15,7 @@ import java.io.IOException; import java.io.InputStream; import java.io.UnsupportedEncodingException; import java.net.Socket; +import java.net.InetSocketAddress; import java.security.NoSuchAlgorithmException; import java.util.Arrays; @@ -174,39 +175,48 @@ public class ProtoCommon { * @param expect_body_len expect response package body length * @return RecvHeaderInfo: errno and pkg body length */ - public static RecvHeaderInfo recvHeader(InputStream in, byte expect_cmd, long expect_body_len) throws IOException { - byte[] header; - int bytes; - long pkg_len; + public static RecvHeaderInfo recvHeader(InputStream in, InetSocketAddress socketAddr, + byte expect_cmd, long expect_body_len) throws IOException + { + byte[] header; + int bytes; + long pkg_len; - header = new byte[FDFS_PROTO_PKG_LEN_SIZE + 2]; - bytes = in.read(header); - if (bytes != header.length) { - if (bytes < 0) { - throw new IOException("connection broken"); - } else { - throw new IOException("recv package size " + bytes + " != " + header.length); - } - } + header = new byte[FDFS_PROTO_PKG_LEN_SIZE + 2]; + bytes = in.read(header); + if (bytes != header.length) { + if (bytes < 0) { + throw new IOException("server " + socketAddr.getAddress().getHostAddress() + + ":" + socketAddr.getPort() + ", connection broken"); + } else { + throw new IOException("server " + socketAddr.getAddress().getHostAddress() + ":" + + socketAddr.getPort() + ", recv package size " + bytes + " != " + header.length); + } + } - if (header[PROTO_HEADER_CMD_INDEX] != expect_cmd) { - throw new IOException("recv cmd: " + header[PROTO_HEADER_CMD_INDEX] + " is not correct, expect cmd: " + expect_cmd); - } + if (header[PROTO_HEADER_CMD_INDEX] != expect_cmd) { + throw new IOException("server " + socketAddr.getAddress().getHostAddress() + ":" + + socketAddr.getPort() + ", recv cmd: " + header[PROTO_HEADER_CMD_INDEX] + + " is not correct, expect cmd: " + expect_cmd); + } - if (header[PROTO_HEADER_STATUS_INDEX] != 0) { - return new RecvHeaderInfo(header[PROTO_HEADER_STATUS_INDEX], 0); - } + if (header[PROTO_HEADER_STATUS_INDEX] != 0) { + return new RecvHeaderInfo(header[PROTO_HEADER_STATUS_INDEX], 0); + } - pkg_len = ProtoCommon.buff2long(header, 0); - if (pkg_len < 0) { - throw new IOException("recv body length: " + pkg_len + " < 0!"); - } + pkg_len = ProtoCommon.buff2long(header, 0); + if (pkg_len < 0) { + throw new IOException("server " + socketAddr.getAddress().getHostAddress() + ":" + + socketAddr.getPort() + ", recv body length: " + pkg_len + " < 0!"); + } - if (expect_body_len >= 0 && pkg_len != expect_body_len) { - throw new IOException("recv body length: " + pkg_len + " is not correct, expect length: " + expect_body_len); - } + if (expect_body_len >= 0 && pkg_len != expect_body_len) { + throw new IOException("server " + socketAddr.getAddress().getHostAddress() + ":" + + socketAddr.getPort() + ", recv body length: " + pkg_len + + " is not correct, expect length: " + expect_body_len); + } - return new RecvHeaderInfo((byte) 0, pkg_len); + return new RecvHeaderInfo((byte)0, pkg_len); } /** @@ -217,36 +227,40 @@ public class ProtoCommon { * @param expect_body_len expect response package body length * @return RecvPackageInfo: errno and reponse body(byte buff) */ - public static RecvPackageInfo recvPackage(InputStream in, byte expect_cmd, long expect_body_len) throws IOException { - RecvHeaderInfo header = recvHeader(in, expect_cmd, expect_body_len); - if (header.errno != 0) { - return new RecvPackageInfo(header.errno, null); - } - - byte[] body = new byte[(int) header.body_len]; - int totalBytes = 0; - int remainBytes = (int) header.body_len; - int bytes = 0; - - while (totalBytes < header.body_len) { - if ((bytes = in.read(body, totalBytes, remainBytes)) < 0) { - break; + public static RecvPackageInfo recvPackage(InputStream in, InetSocketAddress socketAddr, + byte expect_cmd, long expect_body_len) throws IOException + { + RecvHeaderInfo header = recvHeader(in, socketAddr, expect_cmd, expect_body_len); + if (header.errno != 0) { + return new RecvPackageInfo(header.errno, null); } - totalBytes += bytes; - remainBytes -= bytes; - } + byte[] body = new byte[(int) header.body_len]; + int totalBytes = 0; + int remainBytes = (int) header.body_len; + int bytes = 0; - if (totalBytes != header.body_len) { - if (totalBytes == 0) { - throw new IOException("connection broken"); - } else { - throw new IOException("connection broken, recv length: " + totalBytes - + ", expect length: " + header.body_len); - } - } + while (totalBytes < header.body_len) { + if ((bytes = in.read(body, totalBytes, remainBytes)) < 0) { + break; + } - return new RecvPackageInfo((byte) 0, body); + totalBytes += bytes; + remainBytes -= bytes; + } + + if (totalBytes != header.body_len) { + String msg = "server " + socketAddr.getAddress().getHostAddress() + + ":" + socketAddr.getPort() + " connection broken"; + if (totalBytes == 0) { + throw new IOException(msg); + } else { + throw new IOException(msg + ", recv length: " + totalBytes + + ", expect length: " + header.body_len); + } + } + + return new RecvPackageInfo((byte)0, body); } /** @@ -329,7 +343,8 @@ public class ProtoCommon { header = packHeader(FDFS_PROTO_CMD_ACTIVE_TEST, 0, (byte) 0); sock.getOutputStream().write(header); - RecvHeaderInfo headerInfo = recvHeader(sock.getInputStream(), TRACKER_PROTO_CMD_RESP, 0); + InetSocketAddress socketAddr = new InetSocketAddress(sock.getInetAddress(), sock.getPort()); + RecvHeaderInfo headerInfo = recvHeader(sock.getInputStream(), socketAddr, TRACKER_PROTO_CMD_RESP, 0); return headerInfo.errno == 0 ? true : false; } diff --git a/src/main/java/org/csource/fastdfs/StorageClient.java b/src/main/java/org/csource/fastdfs/StorageClient.java index fb808fe..c25dc97 100644 --- a/src/main/java/org/csource/fastdfs/StorageClient.java +++ b/src/main/java/org/csource/fastdfs/StorageClient.java @@ -656,7 +656,7 @@ public class StorageClient { out.write(wholePkg); ProtoCommon.RecvPackageInfo pkgInfo = ProtoCommon.recvPackage(connection.getInputStream(), - ProtoCommon.STORAGE_PROTO_CMD_RESP, -1); + connection.getInetSocketAddress(), ProtoCommon.STORAGE_PROTO_CMD_RESP, -1); this.errno = pkgInfo.errno; if (pkgInfo.errno != 0) { return null; @@ -801,7 +801,7 @@ public class StorageClient { } ProtoCommon.RecvPackageInfo pkgInfo = ProtoCommon.recvPackage(connection.getInputStream(), - ProtoCommon.STORAGE_PROTO_CMD_RESP, -1); + connection.getInetSocketAddress(), ProtoCommon.STORAGE_PROTO_CMD_RESP, -1); this.errno = pkgInfo.errno; if (pkgInfo.errno != 0) { return null; @@ -907,7 +907,7 @@ public class StorageClient { } ProtoCommon.RecvPackageInfo pkgInfo = ProtoCommon.recvPackage(connection.getInputStream(), - ProtoCommon.STORAGE_PROTO_CMD_RESP, 0); + connection.getInetSocketAddress(), ProtoCommon.STORAGE_PROTO_CMD_RESP, 0); this.errno = pkgInfo.errno; if (pkgInfo.errno != 0) { return this.errno; @@ -1004,7 +1004,7 @@ public class StorageClient { } ProtoCommon.RecvPackageInfo pkgInfo = ProtoCommon.recvPackage(connection.getInputStream(), - ProtoCommon.STORAGE_PROTO_CMD_RESP, 0); + connection.getInetSocketAddress(), ProtoCommon.STORAGE_PROTO_CMD_RESP, 0); this.errno = pkgInfo.errno; if (pkgInfo.errno != 0) { return this.errno; @@ -1040,7 +1040,7 @@ public class StorageClient { try { this.send_package(ProtoCommon.STORAGE_PROTO_CMD_DELETE_FILE, group_name, remote_filename, connection); ProtoCommon.RecvPackageInfo pkgInfo = ProtoCommon.recvPackage(connection.getInputStream(), - ProtoCommon.STORAGE_PROTO_CMD_RESP, 0); + connection.getInetSocketAddress(), ProtoCommon.STORAGE_PROTO_CMD_RESP, 0); this.errno = pkgInfo.errno; return pkgInfo.errno; @@ -1122,7 +1122,7 @@ public class StorageClient { out.write(wholePkg); ProtoCommon.RecvPackageInfo pkgInfo = ProtoCommon.recvPackage(connection.getInputStream(), - ProtoCommon.STORAGE_PROTO_CMD_RESP, 0); + connection.getInetSocketAddress(), ProtoCommon.STORAGE_PROTO_CMD_RESP, 0); this.errno = pkgInfo.errno; return pkgInfo.errno; } catch (IOException ex) { @@ -1171,7 +1171,7 @@ public class StorageClient { this.send_download_package(group_name, remote_filename, file_offset, download_bytes, connection); pkgInfo = ProtoCommon.recvPackage(connection.getInputStream(), - ProtoCommon.STORAGE_PROTO_CMD_RESP, -1); + connection.getInetSocketAddress(), ProtoCommon.STORAGE_PROTO_CMD_RESP, -1); this.errno = pkgInfo.errno; if (pkgInfo.errno != 0) { @@ -1232,7 +1232,7 @@ public class StorageClient { this.send_download_package(group_name, remote_filename, file_offset, download_bytes, connection); InputStream in = connection.getInputStream(); - header = ProtoCommon.recvHeader(in, ProtoCommon.STORAGE_PROTO_CMD_RESP, -1); + header = ProtoCommon.recvHeader(in, connection.getInetSocketAddress(), ProtoCommon.STORAGE_PROTO_CMD_RESP, -1); this.errno = header.errno; if (header.errno != 0) { return header.errno; @@ -1320,7 +1320,7 @@ public class StorageClient { this.send_download_package(group_name, remote_filename, file_offset, download_bytes, connection); InputStream in = connection.getInputStream(); - header = ProtoCommon.recvHeader(in, ProtoCommon.STORAGE_PROTO_CMD_RESP, -1); + header = ProtoCommon.recvHeader(in, connection.getInetSocketAddress(), ProtoCommon.STORAGE_PROTO_CMD_RESP, -1); this.errno = header.errno; if (header.errno != 0) { return header.errno; @@ -1376,7 +1376,7 @@ public class StorageClient { this.send_package(ProtoCommon.STORAGE_PROTO_CMD_GET_METADATA, group_name, remote_filename, connection); pkgInfo = ProtoCommon.recvPackage(connection.getInputStream(), - ProtoCommon.STORAGE_PROTO_CMD_RESP, -1); + connection.getInetSocketAddress(), ProtoCommon.STORAGE_PROTO_CMD_RESP, -1); this.errno = pkgInfo.errno; if (pkgInfo.errno != 0) { @@ -1467,7 +1467,7 @@ public class StorageClient { } pkgInfo = ProtoCommon.recvPackage(connection.getInputStream(), - ProtoCommon.STORAGE_PROTO_CMD_RESP, 0); + connection.getInetSocketAddress(), ProtoCommon.STORAGE_PROTO_CMD_RESP, 0); this.errno = pkgInfo.errno; return pkgInfo.errno; @@ -1580,7 +1580,7 @@ public class StorageClient { out.write(wholePkg); pkgInfo = ProtoCommon.recvPackage(connection.getInputStream(), - ProtoCommon.STORAGE_PROTO_CMD_RESP, -1); + connection.getInetSocketAddress(), ProtoCommon.STORAGE_PROTO_CMD_RESP, -1); this.errno = pkgInfo.errno; if (pkgInfo.errno != 0) { diff --git a/src/main/java/org/csource/fastdfs/TrackerClient.java b/src/main/java/org/csource/fastdfs/TrackerClient.java index 83d5ad7..1e2c6ab 100644 --- a/src/main/java/org/csource/fastdfs/TrackerClient.java +++ b/src/main/java/org/csource/fastdfs/TrackerClient.java @@ -118,19 +118,19 @@ public class TrackerClient { } return trackerServer.getConnection(); } catch (IOException e) { - System.err.println("fail over trackerServer get connection error, failOverCount:" + failOverCount + "," + e.getMessage()); + System.err.println("fail over trackerServer get connection error, " + + "failOverCount: " + failOverCount + ", " + e.getMessage()); if (failOverCount == length - 1) { throw e; } } catch (MyException e) { - System.err.println("fail over trackerServer get connection error, failOverCount:" + failOverCount + ", " + e.getMessage()); + System.err.println("fail over trackerServer get connection error, " + + "failOverCount: " + failOverCount + ", " + e.getMessage()); if (failOverCount == length - 1) { throw e; } } - - } return null; } @@ -183,7 +183,7 @@ public class TrackerClient { } ProtoCommon.RecvPackageInfo pkgInfo = ProtoCommon.recvPackage(connection.getInputStream(), - ProtoCommon.TRACKER_PROTO_CMD_RESP, -1); + connection.getInetSocketAddress(), ProtoCommon.TRACKER_PROTO_CMD_RESP, -1); this.errno = pkgInfo.errno; if (pkgInfo.errno != 0) { return null; @@ -275,7 +275,7 @@ public class TrackerClient { } ProtoCommon.RecvPackageInfo pkgInfo = ProtoCommon.recvPackage(connection.getInputStream(), - ProtoCommon.TRACKER_PROTO_CMD_RESP, -1); + connection.getInetSocketAddress(), ProtoCommon.TRACKER_PROTO_CMD_RESP, -1); this.errno = pkgInfo.errno; if (pkgInfo.errno != 0) { return null; @@ -439,7 +439,7 @@ public class TrackerClient { out.write(wholePkg); ProtoCommon.RecvPackageInfo pkgInfo = ProtoCommon.recvPackage(connection.getInputStream(), - ProtoCommon.TRACKER_PROTO_CMD_RESP, -1); + connection.getInetSocketAddress(), ProtoCommon.TRACKER_PROTO_CMD_RESP, -1); this.errno = pkgInfo.errno; if (pkgInfo.errno != 0) { return null; @@ -560,7 +560,7 @@ public class TrackerClient { out.write(header); ProtoCommon.RecvPackageInfo pkgInfo = ProtoCommon.recvPackage(connection.getInputStream(), - ProtoCommon.TRACKER_PROTO_CMD_RESP, -1); + connection.getInetSocketAddress(), ProtoCommon.TRACKER_PROTO_CMD_RESP, -1); this.errno = pkgInfo.errno; if (pkgInfo.errno != 0) { return null; @@ -668,7 +668,7 @@ public class TrackerClient { out.write(wholePkg); ProtoCommon.RecvPackageInfo pkgInfo = ProtoCommon.recvPackage(connection.getInputStream(), - ProtoCommon.TRACKER_PROTO_CMD_RESP, -1); + connection.getInetSocketAddress(), ProtoCommon.TRACKER_PROTO_CMD_RESP, -1); this.errno = pkgInfo.errno; if (pkgInfo.errno != 0) { return null; @@ -749,7 +749,7 @@ public class TrackerClient { out.write(wholePkg); ProtoCommon.RecvPackageInfo pkgInfo = ProtoCommon.recvPackage(connection.getInputStream(), - ProtoCommon.TRACKER_PROTO_CMD_RESP, 0); + connection.getInetSocketAddress(), ProtoCommon.TRACKER_PROTO_CMD_RESP, 0); this.errno = pkgInfo.errno; return pkgInfo.errno == 0; } catch (IOException e) { @@ -893,9 +893,8 @@ public class TrackerClient { System.arraycopy(bs, 0, wholePkg, header.length, bs.length); out.write(wholePkg); - ProtoCommon.RecvPackageInfo pkgInfo = ProtoCommon.recvPackage( - connection.getInputStream(), - ProtoCommon.TRACKER_PROTO_CMD_RESP, -1); + ProtoCommon.RecvPackageInfo pkgInfo = ProtoCommon.recvPackage(connection.getInputStream(), + connection.getInetSocketAddress(), ProtoCommon.TRACKER_PROTO_CMD_RESP, -1); this.errno = pkgInfo.errno; if (pkgInfo.errno != 0) { return null; diff --git a/src/main/java/org/csource/fastdfs/pool/ConnectionManager.java b/src/main/java/org/csource/fastdfs/pool/ConnectionManager.java index 4f39352..a4ecfc3 100644 --- a/src/main/java/org/csource/fastdfs/pool/ConnectionManager.java +++ b/src/main/java/org/csource/fastdfs/pool/ConnectionManager.java @@ -53,7 +53,9 @@ public class ConnectionManager { if (freeCount.get() > 0) { freeCount.decrementAndGet(); connection = freeConnections.poll(); - if (!connection.isAvaliable() || (System.currentTimeMillis() - connection.getLastAccessTime()) > ClientGlobal.g_connection_pool_max_idle_time) { + if (!connection.isAvaliable() || (System.currentTimeMillis() - connection.getLastAccessTime()) > + ClientGlobal.g_connection_pool_max_idle_time) + { closeConnection(connection); continue; } @@ -62,7 +64,8 @@ public class ConnectionManager { try { isActive = connection.activeTest(); } catch (IOException e) { - System.err.println("send to server[" + inetSocketAddress.getAddress().getHostAddress() + ":" + inetSocketAddress.getPort() + "] active test error, emsg: " + e.getMessage()); + System.err.println("send to server " + inetSocketAddress.getAddress().getHostAddress() + + ":" + inetSocketAddress.getPort() + " active test error, emsg: " + e.getMessage()); isActive = false; } if (!isActive) { @@ -72,7 +75,9 @@ public class ConnectionManager { connection.setNeedActiveTest(false); } } - } else if (ClientGlobal.g_connection_pool_max_count_per_entry == 0 || totalCount.get() < ClientGlobal.g_connection_pool_max_count_per_entry) { + } else if (ClientGlobal.g_connection_pool_max_count_per_entry == 0 || totalCount.get() < + ClientGlobal.g_connection_pool_max_count_per_entry) + { connection = ConnectionFactory.create(this.inetSocketAddress); totalCount.incrementAndGet(); } else { @@ -81,10 +86,13 @@ public class ConnectionManager { //wait single success continue; } - throw new MyException("connect to server " + inetSocketAddress.getAddress().getHostAddress() + ":" + inetSocketAddress.getPort() + " fail, wait_time > " + ClientGlobal.g_connection_pool_max_wait_time_in_ms + "ms"); + throw new MyException("connect to server " + inetSocketAddress.getAddress().getHostAddress() + + ":" + inetSocketAddress.getPort() + " fail, wait_time > " + + ClientGlobal.g_connection_pool_max_wait_time_in_ms + "ms"); } catch (InterruptedException e) { e.printStackTrace(); - throw new MyException("connect to server " + inetSocketAddress.getAddress().getHostAddress() + ":" + inetSocketAddress.getPort() + " fail, emsg: " + e.getMessage()); + throw new MyException("connect to server " + inetSocketAddress.getAddress().getHostAddress() + + ":" + inetSocketAddress.getPort() + " fail, emsg: " + e.getMessage()); } } return connection;