You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@ignite.apache.org by "SammyVimes (via GitHub)" <gi...@apache.org> on 2023/07/21 12:05:49 UTC
[GitHub] [ignite-3] SammyVimes commented on a diff in pull request #2326: IGNITE-19009 Introduce file transfer support in messaging
SammyVimes commented on code in PR #2326:
URL: https://github.com/apache/ignite-3/pull/2326#discussion_r1270595756
##########
modules/network/src/main/java/org/apache/ignite/internal/network/direct/stream/DirectByteBufferStreamImplV1.java:
##########
@@ -129,6 +137,18 @@ public class DirectByteBufferStreamImplV1 implements DirectByteBufferStream {
private Iterator<?> it;
+ private BufferedInputStream fileReader;
+
+ private int fileSate = 0;
Review Comment:
```suggestion
private int fileState = 0;
```
##########
modules/network/src/main/java/org/apache/ignite/internal/network/direct/stream/DirectByteBufferStreamImplV1.java:
##########
@@ -129,6 +137,18 @@ public class DirectByteBufferStreamImplV1 implements DirectByteBufferStream {
private Iterator<?> it;
+ private BufferedInputStream fileReader;
+
+ private int fileSate = 0;
+
+ private Path filePath;
+
+ private BufferedOutputStream fileWriter;
+
+ private long bytesWritten = -1;
Review Comment:
```suggestion
private long fileBytesWritten = -1;
```
##########
modules/network/src/main/java/org/apache/ignite/internal/network/direct/stream/DirectByteBufferStreamImplV1.java:
##########
@@ -1662,12 +1682,167 @@ public <C extends Set<?>> C readSet(MessageCollectionItemType itemType, MessageR
return map0;
}
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public void writeFile(File file) {
+ try {
+ if (file == null) {
+ writeInt(-1);
+ return;
+ }
+
+ switch (fileSate) {
+ case 0:
+ writeLong(file.length());
+
+ if (!lastFinished) {
+ return;
+ }
+
+ fileSate++;
+
+ //noinspection fallthrough
+ case 1:
+ writeString(file.getName());
+
+ if (!lastFinished) {
+ return;
+ }
+
+ fileSate++;
+
+ //noinspection fallthrough
+ case 2:
+ lastFinished = false;
+
+ if (fileReader == null) {
+ fileReader = new BufferedInputStream(new FileInputStream(file));
+ }
+
+ int pos = buf.position();
+ int toWrite = Math.min(buf.remaining(), fileReader.available());
+
+ if (toWrite > 0) {
+ byte[] arr = new byte[toWrite];
+ fileReader.read(arr, 0, toWrite);
+ GridUnsafe.copyMemory(arr, BYTE_ARR_OFF, heapArr, baseOff + pos, toWrite);
+ buf.position(pos + toWrite);
+ }
+
+ if (fileReader.available() == 0) {
+ fileReader.close();
+ fileReader = null;
+
+ lastFinished = true;
+
+ fileSate = 0;
+ }
+ break;
+
+ default:
+ throw new IllegalStateException("Invalid file state: " + fileSate);
+ }
+ } catch (IOException e) {
+ if (fileReader != null) {
+ try {
+ fileReader.close();
+ fileReader = null;
+ } catch (IOException ex) {
+ RuntimeException exception = new RuntimeException(ex);
+ exception.addSuppressed(e);
+ throw exception;
+ }
+ }
+ throw new RuntimeException(e);
+ }
+
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public File readFile() {
+ try {
+ switch (fileSate) {
+ case 0:
+ fileSize = readLong();
+ if (!lastFinished) {
+ return null;
+ } else if (fileSize < 0) {
+ return null;
+ }
+
+ fileSate++;
+
+ //noinspection fallthrough
+ case 1:
+ String fileName = readString();
+ if (!lastFinished) {
+ return null;
+ }
+
+ filePath = Files.createTempDirectory("").resolve(fileName);
+ fileWriter = new BufferedOutputStream(new FileOutputStream(filePath.toFile(), true));
+
+ fileSate++;
+ bytesWritten = 0;
+
+ //noinspection fallthrough
+ case 2:
+ lastFinished = false;
+
+ int pos = buf.position();
+ int remaining = buf.remaining();
+ int toRead = (int) Math.min(remaining, fileSize - bytesWritten);
+
+ byte[] arr = new byte[toRead];
Review Comment:
Same here, we should write to writer from bytebuffer without allocations
##########
modules/network/src/main/java/org/apache/ignite/internal/network/direct/stream/DirectByteBufferStreamImplV1.java:
##########
@@ -1662,12 +1682,167 @@ public <C extends Set<?>> C readSet(MessageCollectionItemType itemType, MessageR
return map0;
}
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public void writeFile(File file) {
+ try {
+ if (file == null) {
+ writeInt(-1);
+ return;
+ }
+
+ switch (fileSate) {
+ case 0:
+ writeLong(file.length());
+
+ if (!lastFinished) {
+ return;
+ }
+
+ fileSate++;
+
+ //noinspection fallthrough
+ case 1:
+ writeString(file.getName());
+
+ if (!lastFinished) {
+ return;
+ }
+
+ fileSate++;
+
+ //noinspection fallthrough
+ case 2:
+ lastFinished = false;
+
+ if (fileReader == null) {
+ fileReader = new BufferedInputStream(new FileInputStream(file));
+ }
+
+ int pos = buf.position();
+ int toWrite = Math.min(buf.remaining(), fileReader.available());
+
+ if (toWrite > 0) {
+ byte[] arr = new byte[toWrite];
+ fileReader.read(arr, 0, toWrite);
+ GridUnsafe.copyMemory(arr, BYTE_ARR_OFF, heapArr, baseOff + pos, toWrite);
+ buf.position(pos + toWrite);
+ }
+
+ if (fileReader.available() == 0) {
+ fileReader.close();
+ fileReader = null;
+
+ lastFinished = true;
+
+ fileSate = 0;
+ }
+ break;
+
+ default:
+ throw new IllegalStateException("Invalid file state: " + fileSate);
+ }
+ } catch (IOException e) {
+ if (fileReader != null) {
+ try {
+ fileReader.close();
+ fileReader = null;
+ } catch (IOException ex) {
+ RuntimeException exception = new RuntimeException(ex);
+ exception.addSuppressed(e);
+ throw exception;
+ }
+ }
+ throw new RuntimeException(e);
+ }
+
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public File readFile() {
+ try {
+ switch (fileSate) {
+ case 0:
+ fileSize = readLong();
+ if (!lastFinished) {
+ return null;
+ } else if (fileSize < 0) {
+ return null;
+ }
+
+ fileSate++;
+
+ //noinspection fallthrough
+ case 1:
+ String fileName = readString();
+ if (!lastFinished) {
+ return null;
+ }
+
+ filePath = Files.createTempDirectory("").resolve(fileName);
+ fileWriter = new BufferedOutputStream(new FileOutputStream(filePath.toFile(), true));
Review Comment:
I presume, this doesn't retain file attributes? I'm not sure if they are needed though
##########
modules/network/src/test/resources/files/string.json:
##########
@@ -0,0 +1,5 @@
+{
Review Comment:
Not sure if we need these files, it's easier and cleaner to generate them
##########
modules/network/src/main/java/org/apache/ignite/internal/network/direct/stream/DirectByteBufferStreamImplV1.java:
##########
@@ -1662,12 +1682,167 @@ public <C extends Set<?>> C readSet(MessageCollectionItemType itemType, MessageR
return map0;
}
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public void writeFile(File file) {
+ try {
+ if (file == null) {
+ writeInt(-1);
+ return;
+ }
+
+ switch (fileSate) {
+ case 0:
+ writeLong(file.length());
+
+ if (!lastFinished) {
+ return;
+ }
+
+ fileSate++;
+
+ //noinspection fallthrough
+ case 1:
+ writeString(file.getName());
+
+ if (!lastFinished) {
+ return;
+ }
+
+ fileSate++;
+
+ //noinspection fallthrough
+ case 2:
+ lastFinished = false;
+
+ if (fileReader == null) {
+ fileReader = new BufferedInputStream(new FileInputStream(file));
+ }
+
+ int pos = buf.position();
+ int toWrite = Math.min(buf.remaining(), fileReader.available());
+
+ if (toWrite > 0) {
+ byte[] arr = new byte[toWrite];
Review Comment:
I think it is possible to read file directly to the output buffer to avoid allocations
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org