Add support for sync flush() about zstd-jni HOT 14 CLOSED

I see, makes sense. Will investigate how to provide it.

from zstd-jni.

Can you try with the latest commits on master. I think the problem was not that it has not flushing the internal buffers but because it was not closing the frame. With the current change it will close the frame also if "syncFlush" is true.

from zstd-jni.

BTW, I was just testing it and it look like the syncFlush=false will also work for your case, take a look:

scala> val os = new ZstdOutputStream(new FileOutputStream("/tmp/test.zst"), 1, false) 
os: com.github.luben.zstd.ZstdOutputStream = com.github.luben.zstd.ZstdOutputStream@2c79f93d

scala> val is = new ZstdInputStream(new FileInputStream("/tmp/test.zst"))
is: com.github.luben.zstd.ZstdInputStream = com.github.luben.zstd.ZstdInputStream@32ae5021

scala> val buff = new Array[Byte](8)
buff: Array[Byte] = Array(0, 0, 0, 0, 0, 0, 0, 0)

scala> os.write("foo".getBytes)

scala> os.flush

scala> is.read(buff)
java.io.IOException: Read error or truncated source
  at com.github.luben.zstd.ZstdInputStream.read(ZstdInputStream.java:102)
  at java.io.FilterInputStream.read(FilterInputStream.java:107)
  ... 39 elided

scala> buff
res18: Array[Byte] = Array(102, 111, 111, 0, 0, 0, 0, 0)

scala> os.write("bar".getBytes)

scala> os.flush

scala> is.read(buff)
java.io.IOException: Read error or truncated source
  at com.github.luben.zstd.ZstdInputStream.read(ZstdInputStream.java:102)
  at java.io.FilterInputStream.read(FilterInputStream.java:107)
  ... 39 elided

scala> buff
res22: Array[Byte] = Array(98, 97, 114, 0, 0, 0, 0, 0)

May you should just catch the exception that tells the frame is not finished yet. Another option is to set the InputStream as continuous. Example:

scala> val os = new ZstdOutputStream(new FileOutputStream("/tmp/test.zst"), 1, false) 
os: com.github.luben.zstd.ZstdOutputStream = com.github.luben.zstd.ZstdOutputStream@6b007b5

scala> val is = new ZstdInputStream(new FileInputStream("/tmp/test.zst")).setContinuous(true)
is: com.github.luben.zstd.ZstdInputStream = com.github.luben.zstd.ZstdInputStream@2336f4d5

scala> val buff = new Array[Byte](8)
buff: Array[Byte] = Array(0, 0, 0, 0, 0, 0, 0, 0)

scala> os.write("foo".getBytes)

scala> is.read(buff)
res1: Int = 0

scala> buff
res2: Array[Byte] = Array(0, 0, 0, 0, 0, 0, 0, 0)

scala> os.flush

scala> is.read(buff)
res4: Int = 3

scala> buff
res5: Array[Byte] = Array(102, 111, 111, 0, 0, 0, 0, 0)

scala> os.write("bar".getBytes)

scala> os.flush

scala> is.read(buff)
res8: Int = 3

scala> buff
res9: Array[Byte] = Array(98, 97, 114, 0, 0, 0, 0, 0

scala> is.read(buff)
res10: Int = 0

scala> is.read(buff)
res11: Int = 0

scala> os.close

scala> is.read
res13: Int = 0

scala> is.read
res14: Int = -1

In this case the end will be marked by -1 read size.

from zstd-jni.

Hello, no syncFlush=false is not applicable for me, a i have a thread continuously reading the stream and unpacking messages from it, any type of exceptions or a premature return of read() will cause a full loose of context at a higher level of code and can't be recovered

from zstd-jni.

Got initialize error on master version

Exception in thread "main" java.lang.ExceptionInInitializerError
at IOTest.testZstdCompress(IOTest.java:91)
at IOTest.main(IOTest.java:27)
Caused by: java.lang.UnsupportedOperationException: Unsupported OS/arch, cannot find /linux/amd64/libzstd.so. Please try building from source.
at com.github.luben.zstd.util.Native.load(Native.java:55)
at com.github.luben.zstd.ZstdOutputStream.(ZstdOutputStream.java:17)
... 2 more
t

from zstd-jni.

removing "../../../../../" from resourcePath solves a problem
as i see, this was added to workaround some android bug, maybe it will be better to append this only if os() is android?

from zstd-jni.

Yes. I will revert the path change as it does not solve the android case.

Regarding the issue, aren't you reading in a loop? If you .setContinuous() on the ZstdInputStream does it solve the problem? Currently the only difference is that if syncFlush=true the frame is completed (3 byte added to the stream) and next write will open a new frame.

from zstd-jni.

Imaging the following stream chains:
Output:
DataOutputStream (DOS) -> BufferedOutputStream (BOS) -> ZStdOutputStream (ZOS) -> SocketOutputStream (SOS)
And Input:
DataInputStream (DIS) < BufferedInputStream (BIS) < ZstdInputStream (ZIS) < SocketInputStream (SIS)

now sender calls:
DOS.writeLong(SOME_CMD);
DOS.flush();

and a receiver code:
switch (DIS.readLong()) {....}
calling DIS.readLong() will in turn call DIS.readFully(buf, 0, 8)
DataInputStream.readFully:
public final void readFully(byte b[], int off, int len) throws IOException { if (len < 0) throw new IndexOutOfBoundsException(); int n = 0; while (n < len) { int count = in.read(b, off + n, len - n); if (count < 0) throw new EOFException(); n += count; } }
so readFully() will call BIS.read(b, 0, 8) with will call BIS.fill() calling ZIS.read(b, 0, 8192) - >
ZIS.read(b, 0, 8192) -> SOS.read(b, 0, 8192)
SOS.read will return immediately after ~ 4 (+compress overhead) bytes has been red
(thats how socket input stream works, it returns available data if any or waits for a new data within socket timeout margin)

ZIS will try to decompressStream() in a loop until (srcSize - srcPos == 0)
dstPos < dstSize will never be reached because no such data is available
nor will (size == 0) because a frame is not yet completed
so loop will call SIS.read() again wich will lead to stucked read() forever (until socket timeout)
and this will happen regardless of isContinious flags is set or no

from zstd-jni.

I see. Use then the flushSync=true to finish the frame on .flush(). I will push a revert for the path a little bit later.

from zstd-jni.

With flushSync=true it works almost flawlessly :)
There should be a guard in flush() method against flushing when no data has been written
DeflaterOutputStream does has such a guard
The problem in a DataInputStream again wich considers read() == 0 as EOF
So if on a sender side i do
write();
flush();
flush();

DataInputStream on a receiver side will throw EOFException() while this is wrong

My current workaround to skip empty frames is:

new DataInputStream(
    new BufferedInputStream(
        new ZstdInputStream(sock.getInputStream()) {
            @Override
            public int read(byte[] b, int off, int len) throws IOException {
                int ret;
                do {
                    ret = super.read(b, off, len);
                } while (ret == 0);
                return ret;
            }
        }));

And with InflaterInputStream it is not needed

from zstd-jni.

I see. I will make it not close the frame if nothing is flushed out from zstd buffers

from zstd-jni.

@2x4ever I have made it not close the frame on flush if there is nothing written. It addresses the above case. Also I have reverted the relative path loading on the native library. Is it solving the problem?

from zstd-jni.

Already tried the latest version and it works fantastic!
Many thanks!
Now it is a drop-in replacement for Deflater/Inflater io streams

from zstd-jni.

Great, I will release it with the next version. I should be coming soon.

from zstd-jni.