FiberAsync value created outside of Fiber (but run inside) results in "IllegalThreadStateException: Method called not from within a fiber"

[io7m]

Hello.

The following program:

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import co.paralleluniverse.fibers.Fiber;
import co.paralleluniverse.fibers.FiberAsync;
import co.paralleluniverse.fibers.SuspendExecution;

public final class TestNotFromFiber
{
  private interface IOOpCallbackType<V>
  {
    void onSuccess(
      V v);

    void onFailure(
      IOException e);
  }

  private static final class IOOp implements Runnable
  {
    private final IOOpCallbackType<List<String>> callback;

    IOOp(
      final IOOpCallbackType<List<String>> r)
    {
      this.callback = r;
    }

    @Override public void run()
    {
      try {
        System.out.println("IOOp: running");
        this.callback.onSuccess(Files.readAllLines(
          Paths.get("/etc/passwd"),
          StandardCharsets.UTF_8));
      } catch (final IOException e) {
        this.callback.onFailure(e);
        System.out.println("IOOp: failed");
      } finally {
        System.out.println("IOOp: done");
      }
    }
  }

  private static abstract class IOOpFiberAsync<V> extends
    FiberAsync<V, IOException> implements IOOpCallbackType<V>
  {
    private static final long serialVersionUID = 1L;

    public IOOpFiberAsync()
    {
      super();
    }

    @Override public void onSuccess(
      final V v)
    {
      this.asyncCompleted(v);
    }

    @Override public void onFailure(
      final IOException e)
    {
      this.asyncFailed(e);
    }
  }

  public static void main(
    final String args[])
    throws Exception
  {
    final ExecutorService exec = Executors.newFixedThreadPool(1);

    final IOOpFiberAsync<List<String>> op =
      new IOOpFiberAsync<List<String>>() {
        private static final long serialVersionUID = 1L;

        @Override protected void requestAsync()
        {
          exec.execute(new IOOp(this));
        }
      };

    final Fiber<Void> k = new Fiber<Void>() {
      private static final long serialVersionUID = 1L;

      @Override protected Void run()
        throws SuspendExecution,
          InterruptedException
      {
        try {
          final List<String> r = op.run();
          for (final String l : r) {
            System.out.println("got: " + l);
          }
        } catch (final IOException e) {
          e.printStackTrace();
        }
        return null;
      }
    };

    System.out.println("main: starting");
    k.start();
    System.out.println("main: waiting");
    k.join();
  }
}

Results in:

main: starting
main: waiting
Exception in Fiber "fiber-10000001" java.lang.IllegalThreadStateException: Method called not from within a fiber
    at co.paralleluniverse.fibers.FiberAsync.requestSync(FiberAsync.java:289)
    at co.paralleluniverse.fibers.FiberAsync.runSync(FiberAsync.java:255)
    at co.paralleluniverse.fibers.FiberAsync.run(FiberAsync.java:111)
    at com.io7m.jnecto.tests.catalog.server.quasar.TestNotFromFiber$2.run(TestNotFromFiber.java:99)
    at com.io7m.jnecto.tests.catalog.server.quasar.TestNotFromFiber$2.run(TestNotFromFiber.java:1)
WARNING: Uninstrumented methods (marked '**') or call-sites (marked '!!') detected on the call stack: 
    at co.paralleluniverse.fibers.FiberAsync.requestSync) (FiberAsync.java:289) **
    at co.paralleluniverse.fibers.FiberAsync.runSync (FiberAsync.java:255) **
    at co.paralleluniverse.fibers.FiberAsync.run (FiberAsync.java:111) !! (instrumented suspendable calls at: [])
    at com.io7m.jnecto.tests.catalog.server.quasar.TestNotFromFiber$2.run (TestNotFromFiber.java:99)
    at com.io7m.jnecto.tests.catalog.server.quasar.TestNotFromFiber$2.run (TestNotFromFiber.java:1) (optimized)
    at co.paralleluniverse.fibers.Fiber.run1 (Fiber.java:1014)
Exception in Fiber "fiber-10000001" java.lang.IllegalThreadStateException: Method called not from within a fiber
    at co.paralleluniverse.fibers.FiberAsync.requestSync(FiberAsync.java:289)
    at co.paralleluniverse.fibers.FiberAsync.runSync(FiberAsync.java:255)
    at co.paralleluniverse.fibers.FiberAsync.run(FiberAsync.java:111)
    at com.io7m.jnecto.tests.catalog.server.quasar.TestNotFromFiber$2.run(TestNotFromFiber.java:99)
    at com.io7m.jnecto.tests.catalog.server.quasar.TestNotFromFiber$2.run(TestNotFromFiber.java:1)
Exception in thread "main" java.util.concurrent.ExecutionException: java.lang.IllegalThreadStateException: Method called not from within a fiber
    at co.paralleluniverse.fibers.Fiber.get(Fiber.java:1304)
    at co.paralleluniverse.fibers.Fiber.join(Fiber.java:1279)
    at com.io7m.jnecto.tests.catalog.server.quasar.TestNotFromFiber.main(TestNotFromFiber.java:113)
Caused by: java.lang.IllegalThreadStateException: Method called not from within a fiber
    at co.paralleluniverse.fibers.FiberAsync.requestSync(FiberAsync.java:289)
    at co.paralleluniverse.fibers.FiberAsync.runSync(FiberAsync.java:255)
    at co.paralleluniverse.fibers.FiberAsync.run(FiberAsync.java:111)
    at com.io7m.jnecto.tests.catalog.server.quasar.TestNotFromFiber$2.run(TestNotFromFiber.java:99)
    at com.io7m.jnecto.tests.catalog.server.quasar.TestNotFromFiber$2.run(TestNotFromFiber.java:1)
    at co.paralleluniverse.fibers.Fiber.run1(Fiber.java:1014)
    at co.paralleluniverse.fibers.Fiber.exec(Fiber.java:729)
    at co.paralleluniverse.fibers.FiberForkJoinScheduler$FiberForkJoinTask.exec1(FiberForkJoinScheduler.java:257)
    at co.paralleluniverse.concurrent.forkjoin.ParkableForkJoinTask.doExec(ParkableForkJoinTask.java:116)
    at co.paralleluniverse.concurrent.forkjoin.ParkableForkJoinTask.exec(ParkableForkJoinTask.java:73)
    at jsr166e.ForkJoinTask.doExec(ForkJoinTask.java:261)
    at jsr166e.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:988)
    at jsr166e.ForkJoinPool.runWorker(ForkJoinPool.java:1628)
    at jsr166e.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107)
WARNING: Uninstrumented methods (marked '**') or call-sites (marked '!!') detected on the call stack: 
    at co.paralleluniverse.fibers.FiberAsync.requestSync) (FiberAsync.java:289) **
    at co.paralleluniverse.fibers.FiberAsync.runSync (FiberAsync.java:255) **
    at co.paralleluniverse.fibers.FiberAsync.run (FiberAsync.java:111) !! (instrumented suspendable calls at: [])
    at com.io7m.jnecto.tests.catalog.server.quasar.TestNotFromFiber$2.run (TestNotFromFiber.java:99)
    at com.io7m.jnecto.tests.catalog.server.quasar.TestNotFromFiber$2.run (TestNotFromFiber.java:1) (optimized)
    at co.paralleluniverse.fibers.Fiber.run1 (Fiber.java:1014)

However, moving the op declaration inside the fiber:

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import co.paralleluniverse.fibers.Fiber;
import co.paralleluniverse.fibers.FiberAsync;
import co.paralleluniverse.fibers.SuspendExecution;

public final class TestNotFromFiber
{
  private interface IOOpCallbackType<V>
  {
    void onSuccess(
      V v);

    void onFailure(
      IOException e);
  }

  private static final class IOOp implements Runnable
  {
    private final IOOpCallbackType<List<String>> callback;

    IOOp(
      final IOOpCallbackType<List<String>> r)
    {
      this.callback = r;
    }

    @Override public void run()
    {
      try {
        System.out.println("IOOp: running");
        this.callback.onSuccess(Files.readAllLines(
          Paths.get("/etc/passwd"),
          StandardCharsets.UTF_8));
      } catch (final IOException e) {
        this.callback.onFailure(e);
        System.out.println("IOOp: failed");
      } finally {
        System.out.println("IOOp: done");
      }
    }
  }

  private static abstract class IOOpFiberAsync<V> extends
    FiberAsync<V, IOException> implements IOOpCallbackType<V>
  {
    private static final long serialVersionUID = 1L;

    public IOOpFiberAsync()
    {
      super();
    }

    @Override public void onSuccess(
      final V v)
    {
      this.asyncCompleted(v);
    }

    @Override public void onFailure(
      final IOException e)
    {
      this.asyncFailed(e);
    }
  }

  public static void main(
    final String args[])
    throws Exception
  {
    final ExecutorService exec = Executors.newFixedThreadPool(1);

    final Fiber<Void> k = new Fiber<Void>() {
      private static final long serialVersionUID = 1L;

      @Override protected Void run()
        throws SuspendExecution,
          InterruptedException
      {
        final IOOpFiberAsync<List<String>> op =
          new IOOpFiberAsync<List<String>>() {
            private static final long serialVersionUID = 1L;

            @Override protected void requestAsync()
            {
              exec.execute(new IOOp(this));
            }
          };

        try {
          final List<String> r = op.run();
          for (final String l : r) {
            System.out.println("got: " + l);
          }
        } catch (final IOException e) {
          e.printStackTrace();
        }
        return null;
      }
    };

    System.out.println("main: starting");
    k.start();
    System.out.println("main: waiting");
    k.join();
  }
}

... works correctly.

quasar-core-0.7.2

java version "1.7.0_79"
OpenJDK Runtime Environment (IcedTea 2.5.5) (Arch Linux build 7.u79_2.5.5-1-x86_64)
OpenJDK 64-Bit Server VM (build 24.79-b02, mixed mode)

[circlespainter]

This is by design (for efficiency's sake): a FiberAsync instance will be associated to the invoking fiber at construction time, so it needs to be both created and run by the invoking fiber. The fix commit inserts a note about it in the docs.

[rendaw]

I've spent an hour or two to figure this out, and if I understand correctly the callback the FiberAsync wraps does not need to be run by the invoking fiber - the callback can be run by another thread or by an executor or whatever. I think this is what runBlocking does? I spent the majority of my time looking for some sort of bridging future I could use with a 3rd party thread pool after reading and misunderstanding the above note.

Of course, I might still be misunderstanding.