jvm-streaming 0.3.1 → 0.3.2
raw patch · 4 files changed
+96/−45 lines, 4 filesdep ~inline-javadep ~jnidep ~jvm
Dependency ranges changed: inline-java, jni, jvm, singletons
Files
- jvm-streaming.cabal +7/−5
- src/main/haskell/Language/Java/Streaming.hs +69/−32
- src/test/haskell/Language/Java/StreamingSpec.hs +18/−7
- src/test/haskell/Spec.hs +2/−1
jvm-streaming.cabal view
@@ -1,5 +1,5 @@ name: jvm-streaming-version: 0.3.1+version: 0.3.2 synopsis: Expose Java iterators as streams from the streaming package. description: Please see README.md. homepage: http://github.com/tweag/inline-java/tree/master/jvm-streaming#readme@@ -34,11 +34,11 @@ build-depends: base >= 4.7 && < 5, distributed-closure >= 0.3,- jni >= 0.4.0 && <0.7,- jvm >= 0.4.0 && <0.5,+ jni >= 0.4.0 && <0.8,+ jvm >= 0.4.0 && <0.6, jvm-batching >= 0.1 && <0.2,- inline-java >= 0.8 && <0.9,- singletons >= 2.2,+ inline-java >= 0.9 && <0.10,+ singletons >= 2.6, streaming >= 0.1.4, vector @@ -59,6 +59,8 @@ streaming, text default-language: Haskell2010+ cpp-options: -DHSPEC_DISCOVER=hspec-discover+ benchmark micro-benchmarks type: exitcode-stdio-1.0
src/main/haskell/Language/Java/Streaming.hs view
@@ -19,7 +19,6 @@ {-# OPTIONS_GHC -fno-warn-orphans #-} {-# OPTIONS_GHC -fno-warn-unused-top-binds #-}-{-# OPTIONS_GHC -fplugin=Language.Java.Inline.Plugin #-} module Language.Java.Streaming ( reifyStreamWithBatching@@ -34,11 +33,10 @@ import Data.Proxy import qualified Data.Vector as V import Data.Singletons (SomeSing(..))-import Foreign.Ptr (FunPtr, Ptr, freeHaskellFunPtr, intPtrToPtr, ptrToIntPtr)+import Foreign.Ptr (FunPtr, Ptr, intPtrToPtr, ptrToIntPtr) import Foreign.ForeignPtr.Unsafe (unsafeForeignPtrToPtr) import qualified Foreign.JNI as JNI import qualified Foreign.JNI.Types as JNI-import Foreign.JNI.Types (jnull) import GHC.Stable ( castPtrToStablePtr , castStablePtrToPtr@@ -52,11 +50,12 @@ import Streaming (Bifunctor(first), Stream, Of) import qualified Streaming as Streaming import qualified Streaming.Prelude as Streaming+import System.IO.Unsafe (unsafePerformIO) imports "io.tweag.jvm.batching.*" imports "java.util.Iterator" -type JNIFun a = JNIEnv -> Ptr JObject -> IO a+type JNIFun a = JNIEnv -> Ptr JObject -> Int64 -> IO a foreign import ccall "wrapper" wrapObjectFun :: JNIFun (Ptr (J ty)) -> IO (FunPtr (JNIFun (Ptr (J ty))))@@ -67,37 +66,39 @@ foreign import ccall "&jvm_streaming_freeIterator" freeIteratorPtr :: FunPtr (JNIEnv -> Ptr JObject -> Int64 -> IO ()) -data FunPtrTable = forall ty. FunPtrTable- { nextPtr :: FunPtr (JNIFun (Ptr (J ty)))+data FunPtrTable = FunPtrTable+ { refPtr :: Int64 } freeIterator :: JNIEnv -> Ptr JObject -> Int64 -> IO () freeIterator _ _ ptr = do let sptr = castPtrToStablePtr $ intPtrToPtr $ fromIntegral ptr FunPtrTable{..} <- deRefStablePtr sptr- freeHaskellFunPtr nextPtr+ freeStablePtr $ castPtrToStablePtr $ intPtrToPtr $ fromIntegral refPtr freeStablePtr sptr -- | Reflects a stream with no batching. newIterator :: forall ty. Stream (Of (J ty)) IO () -> IO (J ('Iface "java.util.Iterator" <> '[ty]))-newIterator stream0 = mdo- ref <- newIORef stream0- nextPtr <- wrapObjectFun $ \_ jthis ->- -- Conversion is safe, because result is always a reflected object.- unsafeForeignPtrToPtr <$> Coerce.coerce <$> popStream jthis+newIterator stream0 = do+ ioStreamRef <- newIORef stream0+ refPtr :: Int64 <- fromIntegral . ptrToIntPtr . castStablePtrToPtr <$>+ newStablePtr ioStreamRef -- Keep FunPtr's in a table that can be referenced from the Java side, so -- that they can be freed. tblPtr :: Int64 <- fromIntegral . ptrToIntPtr . castStablePtrToPtr <$> newStablePtr FunPtrTable{..} iterator <- [java| new Iterator() {+ /// A field that the Haskell side sets to true when it reaches the end. private boolean end = false;+ /// Lookahead element - it always points to a valid element unless /// end is true. There is no constructor, so in order to initialize // it, next() must be invoked once. private Object lookahead;+ @Override public boolean hasNext() { return !end; } @@ -105,7 +106,7 @@ public Object next() { if (hasNext()) { final Object temp = lookahead;- lookahead = hsNext();+ lookahead = hsNext($refPtr); return temp; } else throw new java.util.NoSuchElementException();@@ -116,40 +117,76 @@ private native void hsFinalize(long tblPtr); - private native Object hsNext();+ private native Object hsNext(long refPtr); @Override public void finalize() { hsFinalize($tblPtr); } } |]- klass <- JNI.getObjectClass iterator >>= JNI.newGlobalRef+ runOnce $ do+ klass <- JNI.getObjectClass iterator+ registerNativesForIterator klass+ <* JNI.deleteLocalRef klass+ -- Call next once to initialize the iterator.+ () <- [java| { $iterator.next(); } |]+ return $ generic iterator+ where+ -- Given that we always register natives on the same class,+ -- there is no point in registering natives more than once.+ runOnce :: IO a -> IO a+ runOnce action = do+ let {-# NOINLINE ref #-}+ ref = unsafePerformIO $ newIORef Nothing+ readIORef ref >>= \case+ Nothing -> do+ a <- action+ writeIORef ref (Just a)+ return a+ Just a ->+ return a++-- | Registers functions for the native methods of the inner class created in+-- 'newIterator'.+--+-- We keep this helper as a top-level function to ensure that no state tied+-- to a particular iterator leaks in the registered functions. The methods+-- registered here affect all the instances of the inner class.+registerNativesForIterator :: JClass -> IO ()+registerNativesForIterator klass = do fieldEndId <- JNI.getFieldID klass "end" (JNI.signature (sing :: Sing ('Prim "boolean")))- let popStream :: Ptr JObject -> IO (J ty)- popStream ptrThis = do- stream <- readIORef ref- Streaming.uncons stream >>= \case- Nothing -> do- jthis <- JNI.objectFromPtr ptrThis- -- When the stream ends, set the end field to True- -- so the Iterator knows not to call hsNext again.- JNI.setBooleanField jthis fieldEndId 1- return jnull- Just (x, stream') -> do- writeIORef ref stream'- return x+ nextPtr <- wrapObjectFun $ \_ jthis streamRef ->+ -- Conversion is safe, because result is always a reflected object.+ unsafeForeignPtrToPtr . Coerce.coerce <$>+ popStream fieldEndId jthis streamRef JNI.registerNatives klass [ JNI.JNINativeMethod "hsNext"- (methodSignature [] (sing :: Sing ('Class "java.lang.Object")))+ (methodSignature+ [SomeSing (sing :: Sing ('Prim "long"))]+ (sing :: Sing ('Class "java.lang.Object"))+ ) nextPtr , JNI.JNINativeMethod "hsFinalize" (methodSignature [SomeSing (sing :: Sing ('Prim "long"))] (sing :: Sing 'Void)) freeIteratorPtr ]- -- Call next once to initialize the iterator.- () <- [java| { $iterator.next(); } |]- return $ generic iterator+ where+ popStream :: JFieldID -> Ptr JObject -> Int64 -> IO (J ty)+ popStream fieldEndId ptrThis streamRef = do+ let stableRef = castPtrToStablePtr $ intPtrToPtr $ fromIntegral streamRef+ ref <- deRefStablePtr stableRef+ stream <- readIORef ref+ Streaming.uncons stream >>= \case+ Nothing -> do+ jthis <- JNI.objectFromPtr ptrThis+ -- When the stream ends, set the end field to True+ -- so the Iterator knows not to call hsNext again.+ JNI.setBooleanField jthis fieldEndId 1+ return jnull+ Just (x, stream') -> do+ writeIORef ref stream'+ return x -- | Reifies streams from iterators in batches of the given size. reifyStreamWithBatching
src/test/haskell/Language/Java/StreamingSpec.hs view
@@ -1,6 +1,6 @@ {-# LANGUAGE DataKinds #-} {-# LANGUAGE QuasiQuotes #-}-{-# OPTIONS_GHC -fplugin=Language.Java.Inline.Plugin #-}+{-# LANGUAGE ScopedTypeVariables #-} module Language.Java.StreamingSpec where @@ -8,7 +8,7 @@ import Data.IORef import Language.Java import Language.Java.Inline-import Language.Java.Streaming ()+import Language.Java.Streaming import Test.Hspec import Streaming (Stream, Of) import qualified Streaming.Prelude as Streaming@@ -19,26 +19,37 @@ it "succeeds on empty lists" $ do vals <- reflect [1..0 :: Int32] iterator <- [java| java.util.Arrays.asList($vals).iterator() |]- stream <- reify iterator+ stream :: Stream (Of Int32) IO () <- reify iterator Streaming.toList_ stream `shouldReturn` [1..0 :: Int32] it "succeeds on singleton lists" $ do vals <- reflect [1 :: Int32] iterator <- [java| java.util.Arrays.asList($vals).iterator() |]- stream <- reify iterator+ stream :: Stream (Of Int32) IO () <- reify iterator Streaming.toList_ stream `shouldReturn` [1 :: Int32] it "succeeds on non-trivial lists" $ do vals <- reflect [1..10000 :: Int32] iterator <- [java| java.util.Arrays.asList($vals).iterator() |]- stream <- reify iterator+ stream :: Stream (Of Int32) IO () <- reify iterator Streaming.toList_ stream `shouldReturn` [1..10000 :: Int32] describe "streams" $ do it "have the property that reify . reflect == id" $ do iterator <- reflect (Streaming.each [1..10000] :: Stream (Of Int32) IO ())- stream <- reify iterator+ stream :: Stream (Of Int32) IO () <- reify iterator Streaming.toList_ stream `shouldReturn` [1..10000 :: Int32] it "don't redo effects" $ do ref <- newIORef (1 :: Int32) iterator <- reflect $ Streaming.replicateM 10 $ atomicModifyIORef ref (\i -> (i + 1, i))- stream <- reify iterator+ stream :: Stream (Of Int32) IO () <- reify iterator Streaming.toList_ stream `shouldReturn` [1..10 :: Int32]+ it "are independent one from another" $ do+ ref <- newIORef (1 :: Int32)+ iterator <- reflectStreamWithBatching 1 $+ Streaming.replicateM 10 $ atomicModifyIORef ref (\i -> (i + 1, i))+ ref2 <- newIORef (21 :: Int32)+ iterator2 <- reflectStreamWithBatching 1 $+ Streaming.replicateM 10 $ atomicModifyIORef ref2 (\i -> (i + 1, i))+ stream :: Stream (Of Int32) IO () <- reifyStreamWithBatching 1 iterator+ stream2 :: Stream (Of Int32) IO () <- reifyStreamWithBatching 1 iterator2+ Streaming.toList_ stream `shouldReturn` [1..10 :: Int32]+ Streaming.toList_ stream2 `shouldReturn` [21..30 :: Int32]
src/test/haskell/Spec.hs view
@@ -1,1 +1,2 @@-{-# OPTIONS_GHC -F -pgmF hspec-discover -optF --module-name=Spec #-}+{-# LANGUAGE CPP #-}+{-# OPTIONS_GHC -F -pgmF HSPEC_DISCOVER -optF --module-name=Spec #-}