You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@ignite.apache.org by "Flaugh24 (via GitHub)" <gi...@apache.org> on 2023/07/17 18:16:01 UTC
[GitHub] [ignite-3] Flaugh24 opened a new pull request, #2326: IGNITE-19009 WIP
Flaugh24 opened a new pull request, #2326:
URL: https://github.com/apache/ignite-3/pull/2326
https://issues.apache.org/jira/browse/IGNITE-19009
--
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
[GitHub] [ignite-3] Flaugh24 closed pull request #2326: IGNITE-19009 Introduce file transfer support in messaging
Posted by "Flaugh24 (via GitHub)" <gi...@apache.org>.
Flaugh24 closed pull request #2326: IGNITE-19009 Introduce file transfer support in messaging
URL: https://github.com/apache/ignite-3/pull/2326
--
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
[GitHub] [ignite-3] SammyVimes commented on a diff in pull request #2326: IGNITE-19009 Introduce file transfer support in messaging
Posted by "SammyVimes (via GitHub)" <gi...@apache.org>.
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
[GitHub] [ignite-3] PakhomovAlexander commented on a diff in pull request #2326: IGNITE-19009 Introduce file transfer support in messaging
Posted by "PakhomovAlexander (via GitHub)" <gi...@apache.org>.
PakhomovAlexander commented on code in PR #2326:
URL: https://github.com/apache/ignite-3/pull/2326#discussion_r1272425796
##########
modules/network/src/integrationTest/java/org/apache/ignite/network/scalecube/ItScaleCubeNetworkMessagingTest.java:
##########
@@ -329,6 +339,42 @@ public void testMessageGroupsHandlers(TestInfo testInfo) throws Exception {
assertThat(networkMessageFuture, willBe(equalTo(networkMessage)));
}
+ /**
+ * Tests that correct files are sent and received.
+ *
+ * @throws Exception in case of errors.
+ */
+ @Test
+ public void testBadFileSequence(TestInfo testInfo) throws Exception {
+ testCluster = new Cluster(2, testInfo);
+ testCluster.startAwait();
+
+ ClusterService node1 = testCluster.members.get(0);
+ ClusterService node2 = testCluster.members.get(1);
+
+ List<NetworkMessage> messages = new CopyOnWriteArrayList<>();
+
+ node1.messagingService().addMessageHandler(
+ TestMessageTypes.class,
+ (message, sender, correlationId) -> messages.add(message)
+ );
+
+ // send a sequence of messages, one of which is a file message with a non-existent file.
+ FileMessage file1 = messageFactory.fileMessage().file(randomFile(100)).build();
+ FileMessage nonExistingFile = messageFactory.fileMessage().file(new File("non-existent-file")).build();
+ FileMessage file2 = messageFactory.fileMessage().file(randomFile(200)).build();
+
+ assertThat(node2.messagingService().send(node1.topologyService().localMember(), file1), willCompleteSuccessfully());
+ assertThat(
+ node2.messagingService().send(node1.topologyService().localMember(), nonExistingFile),
+ willThrow(RuntimeException.class, "non-existent-file")
Review Comment:
In this test class I can see the code: `ExecutionException e = assertThrows(ExecutionException.class, () -> invoke0.get(1, TimeUnit.SECONDS));` Do we specify the exceptions we throw from messaging service?
Here we throw `RuntimeException` but in other cases, we throw `CompletionException`. Also, why is it `RuntimeException`? Can it be a nio-related exception since we work with File API?
##########
modules/network/src/integrationTest/java/org/apache/ignite/network/scalecube/ItScaleCubeNetworkMessagingTest.java:
##########
@@ -329,6 +339,42 @@ public void testMessageGroupsHandlers(TestInfo testInfo) throws Exception {
assertThat(networkMessageFuture, willBe(equalTo(networkMessage)));
}
+ /**
+ * Tests that correct files are sent and received.
+ *
+ * @throws Exception in case of errors.
+ */
+ @Test
+ public void testBadFileSequence(TestInfo testInfo) throws Exception {
+ testCluster = new Cluster(2, testInfo);
+ testCluster.startAwait();
+
+ ClusterService node1 = testCluster.members.get(0);
+ ClusterService node2 = testCluster.members.get(1);
+
+ List<NetworkMessage> messages = new CopyOnWriteArrayList<>();
+
+ node1.messagingService().addMessageHandler(
+ TestMessageTypes.class,
+ (message, sender, correlationId) -> messages.add(message)
+ );
+
+ // send a sequence of messages, one of which is a file message with a non-existent file.
+ FileMessage file1 = messageFactory.fileMessage().file(randomFile(100)).build();
+ FileMessage nonExistingFile = messageFactory.fileMessage().file(new File("non-existent-file")).build();
+ FileMessage file2 = messageFactory.fileMessage().file(randomFile(200)).build();
+
+ assertThat(node2.messagingService().send(node1.topologyService().localMember(), file1), willCompleteSuccessfully());
+ assertThat(
+ node2.messagingService().send(node1.topologyService().localMember(), nonExistingFile),
+ willThrow(RuntimeException.class, "non-existent-file")
+ );
+ assertThat(node2.messagingService().send(node1.topologyService().localMember(), file2), willCompleteSuccessfully());
+
+ // make sure that the correct files were delivered.
+ await().until(() -> messages.size() == 2);
Review Comment:
This condition does not check that the correct files were delivered. It checks that there will be at least two messages sooner or later. One of those messages can be the message from the incorrect file. I belive there is a better way to check what we really want to check.
##########
modules/network/src/main/java/org/apache/ignite/internal/network/direct/stream/DirectByteBufferStreamImplV1.java:
##########
@@ -1662,12 +1681,146 @@ 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 (fileState) {
+ case 0:
+ writeLong(file.length());
+
+ if (!lastFinished) {
+ return;
+ }
+
+ fileState++;
+
+ //noinspection fallthrough
+ case 1:
+ writeString(file.getName());
+
+ if (!lastFinished) {
+ return;
+ }
+
+ fileState++;
+
+ //noinspection fallthrough
+ case 2:
+ lastFinished = false;
+
+ if (readFileChannel == null) {
+ readFileChannel = FileChannel.open(file.toPath(), StandardOpenOption.READ);
+ }
+
+ readFileChannel.read(buf);
+
+ if (readFileChannel.position() == readFileChannel.size()) {
+ IgniteUtils.closeAll(readFileChannel);
+ readFileChannel = null;
+
+ lastFinished = true;
+ fileState = 0;
+ }
+ break;
+
+ default:
+ throw new IllegalStateException("Invalid file state: " + fileState);
+ }
+ } catch (Exception e) {
+ try {
+ IgniteUtils.closeAll(readFileChannel);
+ readFileChannel = null;
+ } catch (Exception ex) {
+ ex.addSuppressed(e);
+ throw new RuntimeException(ex);
+ }
+ throw new RuntimeException(e);
+ }
+
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public File readFile() {
+ try {
+ switch (fileState) {
+ case 0:
+ fileSize = readLong();
+ if (!lastFinished) {
+ return null;
+ } else if (fileSize < 0) {
+ return null;
+ }
+
+ fileState++;
+
+ //noinspection fallthrough
+ case 1:
+ String fileName = readString();
+ if (!lastFinished) {
+ return null;
+ }
+
+ filePath = Files.createTempDirectory("").resolve(fileName);
+ writeFileChannel = FileChannel.open(filePath, StandardOpenOption.CREATE, StandardOpenOption.APPEND);
+
+ fileState++;
+ fileBytesWritten = 0;
+
+ //noinspection fallthrough
+ case 2:
+ lastFinished = false;
+
+ if (buf.hasRemaining()) {
+ int toRead = (int) Math.min(buf.remaining(), fileSize - fileBytesWritten);
+ buf.limit(buf.position() + toRead);
+ writeFileChannel.write(buf);
+ fileBytesWritten += toRead;
+ }
+
+ if (fileBytesWritten == fileSize) {
+ IgniteUtils.closeAll(writeFileChannel);
+ writeFileChannel = null;
+
+ lastFinished = true;
+ fileState = 0;
+ fileBytesWritten = -1;
+
+ return filePath.toFile();
+ } else {
+ return null;
+ }
+
+ default:
+ throw new IllegalArgumentException("Unknown fileState: " + fileState);
+ }
+ } catch (Exception e) {
+ try {
+ IgniteUtils.closeAll(writeFileChannel);
Review Comment:
should we also `writeFileChannel = null;` here?
--
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
[GitHub] [ignite-3] Flaugh24 commented on a diff in pull request #2326: IGNITE-19009 Introduce file transfer support in messaging
Posted by "Flaugh24 (via GitHub)" <gi...@apache.org>.
Flaugh24 commented on code in PR #2326:
URL: https://github.com/apache/ignite-3/pull/2326#discussion_r1270626847
##########
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:
Purpose of this patch it avoids collecting the whole content of the file in a heap (byte array). We can support attributes later if we need them. But right now, we don't need them.
--
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
[GitHub] [ignite-3] Flaugh24 commented on a diff in pull request #2326: IGNITE-19009 Introduce file transfer support in messaging
Posted by "Flaugh24 (via GitHub)" <gi...@apache.org>.
Flaugh24 commented on code in PR #2326:
URL: https://github.com/apache/ignite-3/pull/2326#discussion_r1270712311
##########
modules/network/src/test/resources/files/string.json:
##########
@@ -0,0 +1,5 @@
+{
Review Comment:
Removed empty and JSON files. But I would prefer to keep the jar file and the png image - the first one is our main case of file transferring (deployment units), and the second one is just to check that we don't serialize files into strings. It's quite hard to generate them.
--
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