[GR-14763] Implement Internal gzcon.

PullRequest: fastr/1978

Author: tomasstupka

com.oracle.truffle.r.ffi.impl/src/com/oracle/truffle/r/ffi/impl/common/JavaUpCallsRFFIImpl.java

@@ -1510,7 +1510,7 @@ public Object getSummaryDescription(Object x) {
     @TruffleBoundary
     public Object getConnectionClassString(Object x) {
         BaseRConnection conn = guaranteeInstanceOf(x, BaseRConnection.class);
-        return wrapString(conn.getConnectionClass());
+        return wrapString(conn.getConnectionClassName());
     }
 
     @Override

com.oracle.truffle.r.nodes.builtin/src/com/oracle/truffle/r/nodes/builtin/base/BasePackage.java

@@ -580,6 +580,7 @@ public BasePackage() {
         add(GrepFunctions.Regexec.class, GrepFunctionsFactory.RegexecNodeGen::create);
         add(GrepFunctions.Strsplit.class, GrepFunctionsFactory.StrsplitNodeGen::create);
         add(GrepFunctions.Sub.class, GrepFunctionsFactory.SubNodeGen::create);
+        add(GZCon.class, GZConNodeGen::create);
         add(HiddenInternalFunctions.GetRegisteredRoutines.class, HiddenInternalFunctionsFactory.GetRegisteredRoutinesNodeGen::create);
         add(HiddenInternalFunctions.ImportIntoEnv.class, HiddenInternalFunctionsFactory.ImportIntoEnvNodeGen::create);
         add(HiddenInternalFunctions.LazyLoadDBFetch.class, HiddenInternalFunctionsFactory.LazyLoadDBFetchNodeGen::create);

com.oracle.truffle.r.nodes.builtin/src/com/oracle/truffle/r/nodes/builtin/base/ConnectionFunctions.java

@@ -506,7 +506,7 @@ protected RList summary(int object) {
             BaseRConnection baseCon = RConnection.fromIndex(object);
             Object[] data = new Object[NAMES.getLength()];
             data[0] = baseCon.getSummaryDescription();
-            data[1] = baseCon.getConnectionClass();
+            data[1] = baseCon.getConnectionClassName();
             data[2] = baseCon.getOpenMode().summaryString();
             data[3] = baseCon.isTextMode() ? "text" : "binary";
             data[4] = baseCon.isOpen() ? "opened" : "closed";

com.oracle.truffle.r.nodes.builtin/src/com/oracle/truffle/r/nodes/builtin/base/GZCon.java

@@ -0,0 +1,70 @@
+/*
+ * Copyright (c) 2014, 2019, Oracle and/or its affiliates. All rights reserved.
+ * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
+ *
+ * This code is free software; you can redistribute it and/or modify it
+ * under the terms of the GNU General Public License version 3 only, as
+ * published by the Free Software Foundation.
+ *
+ * This code is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License
+ * version 3 for more details (a copy is included in the LICENSE file that
+ * accompanied this code).
+ *
+ * You should have received a copy of the GNU General Public License version
+ * 3 along with this work; if not, write to the Free Software Foundation,
+ * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
+ *
+ * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA
+ * or visit www.oracle.com if you need additional information or have any
+ * questions.
+ */
+package com.oracle.truffle.r.nodes.builtin.base;
+
+import com.oracle.truffle.api.dsl.Cached;
+import com.oracle.truffle.api.dsl.Specialization;
+import com.oracle.truffle.api.profiles.ConditionProfile;
+import static com.oracle.truffle.r.nodes.builtin.CastBuilder.Predef.toBoolean;
+import com.oracle.truffle.r.nodes.builtin.RBuiltinNode;
+import static com.oracle.truffle.r.runtime.RCompression.Type.GZIP;
+import com.oracle.truffle.r.runtime.RError;
+import static com.oracle.truffle.r.runtime.builtins.RBehavior.PURE;
+import com.oracle.truffle.r.runtime.builtins.RBuiltin;
+import static com.oracle.truffle.r.runtime.builtins.RBuiltinKind.INTERNAL;
+import com.oracle.truffle.r.runtime.conn.ConnectionSupport.BaseRConnection;
+import com.oracle.truffle.r.runtime.conn.ConnectionSupport.ConnectionClass;
+import static com.oracle.truffle.r.runtime.conn.ConnectionSupport.ConnectionClass.GZCon;
+import com.oracle.truffle.r.runtime.conn.RConnection;
+import com.oracle.truffle.r.runtime.data.model.RAbstractIntVector;
+import java.io.IOException;
+
+@RBuiltin(name = "gzcon", kind = INTERNAL, parameterNames = {"con", "level", "allowNonCompressed", "text"}, behavior = PURE)
+public abstract class GZCon extends RBuiltinNode.Arg4 {
+
+    static {
+        Casts casts = new Casts(GZCon.class);
+        casts.arg("con").defaultError(RError.Message.NOT_CONNECTION, "con").mustNotBeNull().asIntegerVector().findFirst();
+        casts.arg("level").asIntegerVector().findFirst();
+        casts.arg("allowNonCompressed").asLogicalVector().findFirst().map(toBoolean());
+        casts.arg("text").asLogicalVector().findFirst().map(toBoolean());
+    }
+
+    @Specialization
+    public RAbstractIntVector gzcon(int conIndex, @SuppressWarnings("unused") int level, @SuppressWarnings("unused") boolean allowNonCompressed, @SuppressWarnings("unused") boolean text,
+                    @Cached("createBinaryProfile()") ConditionProfile gzConProfile) {
+        BaseRConnection base = RConnection.fromIndex(conIndex);
+        if (gzConProfile.profile(base.getConnectionClass() == ConnectionClass.GZCon)) {
+            RError.warning(this, RError.Message.IS_GZCON);
+            return base.asVector();
+        }
+        try {
+            base.setCompressiontype(GZIP);
+            base.updateConnectionClass(GZCon);
+        } catch (IOException ex) {
+            throw error(RError.Message.GENERIC, ex.getMessage());
+        }
+        return base.asVector();
+    }
+
+}

com.oracle.truffle.r.runtime/src/com/oracle/truffle/r/runtime/RError.java

@@ -1045,7 +1045,8 @@ public enum Message {
         DATE_TIME_CONVERSION_SPEC_NOT_IMPLEMENTED("Date time conversion format '%s' is not implemented in FastR yet. Please submit an issue at https://github.com/oracle/fastr."),
         CANNOT_ALLOCATE_VECTOR_GB("cannot allocate vector of size %.1f Gb"),
         INVALID_POLYNOMIAL_COEFFICIENT("invalid polynomial coefficient"),
-        ROOT_FINDING_FAILED("root finding code failed");
+        ROOT_FINDING_FAILED("root finding code failed"),
+        IS_GZCON("this is already a 'gzcon' connection");
 
         public final String message;
         final boolean hasArgs;

com.oracle.truffle.r.runtime/src/com/oracle/truffle/r/runtime/conn/ConnectionSupport.java

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2014, 2018, Oracle and/or its affiliates. All rights reserved.
+ * Copyright (c) 2014, 2019, Oracle and/or its affiliates. All rights reserved.
  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
  *
  * This code is free software; you can redistribute it and/or modify it
@@ -41,6 +41,7 @@
 
 import com.oracle.truffle.api.CompilerDirectives.TruffleBoundary;
 import com.oracle.truffle.api.object.DynamicObject;
+import com.oracle.truffle.r.runtime.RCompression;
 import com.oracle.truffle.r.runtime.RError;
 import com.oracle.truffle.r.runtime.RError.Message;
 import com.oracle.truffle.r.runtime.RInternalError;
@@ -317,6 +318,7 @@ public enum ConnectionClass {
         Terminal("terminal"),
         File("file"),
         GZFile("gzfile"),
+        GZCon("gzcon"),
         BZFile("bzfile"),
         XZFile("xzfile"),
         Socket("sockconn"),
@@ -641,7 +643,11 @@ private BaseRConnection(ConnectionClass conClass, OpenMode mode) {
             this.openMode = mode;
         }
 
-        public String getConnectionClass() {
+        public ConnectionClass getConnectionClass() {
+            return conClass;
+        }
+
+        public String getConnectionClassName() {
             return conClass.getPrintName();
         }
 
@@ -743,8 +749,12 @@ protected void checkOpen() {
         }
 
         protected void setDelegate(DelegateRConnection conn) {
+            setDelegate(conn, true);
+        }
+
+        protected void setDelegate(DelegateRConnection conn, boolean opened) {
             this.theConnection = conn;
-            opened = true;
+            this.opened = opened;
         }
 
         protected String[] readLinesInternal(int n, EnumSet<ReadLineWarning> warn, boolean skipNul) throws IOException {
@@ -865,6 +875,10 @@ public boolean isSeekable() {
          */
         protected abstract void createDelegateConnection() throws IOException;
 
+        public void setCompressiontype(@SuppressWarnings("unused") RCompression.Type cType) throws IOException {
+            throw new IOException();
+        }
+
         /**
          * Return the value that is used in the "description" field by {@code summary.connection}.
          */
@@ -1061,7 +1075,7 @@ protected void setIncomplete(boolean b) {
         }
 
         public final RAbstractIntVector asVector() {
-            String[] classes = new String[]{ConnectionSupport.getBaseConnection(this).getConnectionClass(), "connection"};
+            String[] classes = new String[]{ConnectionSupport.getBaseConnection(this).getConnectionClassName(), "connection"};
 
             RAbstractIntVector result = RDataFactory.createIntVector(new int[]{getDescriptor()}, true);
 

com.oracle.truffle.r.runtime/src/com/oracle/truffle/r/runtime/conn/DelegateRConnection.java

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2014, 2018, Oracle and/or its affiliates. All rights reserved.
+ * Copyright (c) 2014, 2019, Oracle and/or its affiliates. All rights reserved.
  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
  *
  * This code is free software; you can redistribute it and/or modify it
@@ -47,6 +47,8 @@
 import com.oracle.truffle.r.runtime.data.RDataFactory;
 import com.oracle.truffle.r.runtime.data.RObject;
 import com.oracle.truffle.r.runtime.data.model.RAbstractStringVector;
+import java.util.zip.GZIPInputStream;
+import java.util.zip.GZIPOutputStream;
 
 /**
  * Actually performs the I/O operations for a connections.<br>
@@ -628,4 +630,81 @@ public void closeAndDestroy() throws IOException {
         base.closed = true;
         close();
     }
+
+    private static final int GZIP_BUFFER_SIZE = (2 << 20);
+
+    static DelegateRConnection createGZIPDelegateOutputConnection(BaseRConnection base, OutputStream os) throws IOException {
+        assert base.getOpenMode().canWrite();
+        return new CompressedOutputRConnection(base, new GZIPOutputStream(os, GZIP_BUFFER_SIZE), true);
+    }
+
+    static DelegateRConnection createGZIPDelegateInputConnection(BaseRConnection base, InputStream is) throws IOException {
+        assert base.getOpenMode().canRead();
+        return new CompressedInputRConnection(base, new GZIPInputStream(is, GZIP_BUFFER_SIZE));
+    }
+
+    static class CompressedOutputRConnection extends DelegateWriteRConnection {
+        protected ByteChannel channel;
+        private final boolean seekable;
+        private long seekPosition = 0L;
+
+        protected CompressedOutputRConnection(BaseRConnection base, OutputStream os, boolean seekable) {
+            super(base);
+            this.seekable = seekable;
+            this.channel = ConnectionSupport.newChannel(os);
+        }
+
+        @Override
+        public void closeAndDestroy() throws IOException {
+            base.closed = true;
+            close();
+        }
+
+        @Override
+        protected long seekInternal(long offset, RConnection.SeekMode seekMode, RConnection.SeekRWMode seekRWMode) throws IOException {
+            if (seekable) {
+                // TODO GZIP is basically seekable; however, the output stream does not allow any
+                // seeking
+                long oldPos = seekPosition;
+                seekPosition = offset;
+                return oldPos;
+            }
+            return super.seek(offset, seekMode, seekRWMode);
+        }
+
+        @Override
+        public boolean isSeekable() {
+            return seekable;
+        }
+
+        @Override
+        public ByteChannel getChannel() {
+            return channel;
+        }
+
+        @Override
+        public void truncate() throws IOException {
+            throw RError.nyi(RError.SHOW_CALLER, "truncating compressed file not");
+        }
+    }
+
+    static class CompressedInputRConnection extends DelegateReadRConnection {
+        private final ByteChannel channel;
+
+        protected CompressedInputRConnection(BaseRConnection base, InputStream is) {
+            super(base);
+            channel = ConnectionSupport.newChannel(is);
+        }
+
+        @Override
+        public ByteChannel getChannel() {
+            return channel;
+        }
+
+        @Override
+        public boolean isSeekable() {
+            return false;
+        }
+    }
+
 }