Add support for channel-based breakpoints

CHANGELOG.md

@@ -33,6 +33,8 @@
  - Added CSP version of PingPong benchmark.
 
  - Added simple STM implementation. See `s.i.t.Transactions` and [PR #81](https://github.com/smarr/SOMns/pull/81) for details.
+
+ - Added breakpoints for channel operations in [PR #99](https://github.com/smarr/SOMns/pull/81).
 
 ## 0.1.0 - 2016-12-15
 

src/som/VM.java

@@ -39,7 +39,7 @@
 import som.vm.VmSettings;
 import som.vm.constants.KernelObj;
 import som.vmobjects.SObjectWithClass.SObjectWithoutFields;
-import tools.actors.ActorExecutionTrace;
+import tools.concurrency.ActorExecutionTrace;
 import tools.debugger.Tags;
 import tools.debugger.WebDebugger;
 import tools.dym.DynamicMetrics;

src/som/interpreter/SomLanguage.java

@@ -22,7 +22,10 @@
 import som.vm.NotYetImplementedException;
 import som.vm.constants.Nil;
 import som.vmobjects.SClass;
-import tools.actors.Tags.EventualMessageSend;
+import tools.concurrency.Tags.ChannelRead;
+import tools.concurrency.Tags.ChannelWrite;
+import tools.concurrency.Tags.EventualMessageSend;
+import tools.concurrency.Tags.ExpressionBreakpoint;
 import tools.debugger.Tags.ArgumentTag;
 import tools.debugger.Tags.CommentTag;
 import tools.debugger.Tags.DelimiterClosingTag;
@@ -82,7 +85,8 @@
   StringAccess.class, OpClosureApplication.class, OpArithmetic.class,
   OpComparison.class, OpLength.class,
 
-  EventualMessageSend.class
+  EventualMessageSend.class, ChannelRead.class, ChannelWrite.class,
+  ExpressionBreakpoint.class
 })
 public final class SomLanguage extends TruffleLanguage<VM> {
 

src/som/interpreter/actors/Actor.java

@@ -20,7 +20,7 @@
 import som.vmobjects.SArray.STransferArray;
 import som.vmobjects.SObject;
 import som.vmobjects.SObjectWithClass.SObjectWithoutFields;
-import tools.actors.ActorExecutionTrace;
+import tools.concurrency.ActorExecutionTrace;
 import tools.debugger.WebDebugger;
 
 

src/som/interpreter/actors/EventualSendNode.java

@@ -31,7 +31,8 @@
 import som.vmobjects.SSymbol;
 import tools.SourceCoordinate;
 import tools.SourceCoordinate.FullSourceCoordinate;
-import tools.actors.Tags.EventualMessageSend;
+import tools.concurrency.Tags.EventualMessageSend;
+import tools.concurrency.Tags.ExpressionBreakpoint;
 import tools.debugger.nodes.AbstractBreakpointNode;
 import tools.debugger.nodes.BreakpointNodeGen;
 import tools.debugger.nodes.DisabledBreakpointNode;
@@ -289,7 +290,7 @@ public final Object toNearRefWithoutResultPromise(final Object[] args) {
 
     @Override
     protected boolean isTaggedWith(final Class<?> tag) {
-      if (tag == EventualMessageSend.class) {
+      if (tag == EventualMessageSend.class || tag == ExpressionBreakpoint.class) {
         return true;
       }
       return super.isTaggedWith(tag);

src/som/interpreter/actors/SPromise.java

@@ -17,7 +17,7 @@
 import som.vmobjects.SBlock;
 import som.vmobjects.SClass;
 import som.vmobjects.SObjectWithClass;
-import tools.actors.ActorExecutionTrace;
+import tools.concurrency.ActorExecutionTrace;
 
 
 public class SPromise extends SObjectWithClass {

src/som/interpreter/processes/SChannel.java

@@ -12,10 +12,19 @@ public class SChannel extends SAbstractObject {
   public final SChannelOutput out;
   public final SChannelInput  in;
 
+  /** Indicate that a breakpoint on the writer requested a suspension on read. */
+  private volatile boolean breakAfterRead;
+
+  /** Indicate that a breakpoint on the reader requested a suspension on write. */
+  private volatile boolean breakAfterWrite;
+
   public SChannel() {
+    breakAfterRead  = false;
+    breakAfterWrite = false;
+
     SynchronousQueue<Object> cell = new SynchronousQueue<>();
-    out = new SChannelOutput(cell);
-    in  = new SChannelInput(cell);
+    out = new SChannelOutput(cell, this);
+    in  = new SChannelInput(cell, this);
   }
 
   @Override
@@ -31,15 +40,27 @@ public boolean isValue() {
 
   public static final class SChannelInput extends SAbstractObject {
     private final SynchronousQueue<Object> cell;
+    private final SChannel channel;
 
-    public SChannelInput(final SynchronousQueue<Object> cell) {
-      this.cell = cell;
+    public SChannelInput(final SynchronousQueue<Object> cell,
+        final SChannel channel) {
+      this.cell    = cell;
+      this.channel = channel;
     }
 
     public Object read() throws InterruptedException {
       return cell.take();
     }
 
+    public Object readAndSuspendWriter(final boolean doSuspend) throws InterruptedException {
+      channel.breakAfterWrite = doSuspend;
+      return read();
+    }
+
+    public boolean shouldBreakAfterRead() {
+      return channel.breakAfterRead;
+    }
+
     @Override
     public SClass getSOMClass() {
       assert ChannelPrimitives.In != null;
@@ -54,15 +75,27 @@ public boolean isValue() {
 
   public static final class SChannelOutput extends SAbstractObject {
     private final SynchronousQueue<Object> cell;
+    private final SChannel channel;
 
-    public SChannelOutput(final SynchronousQueue<Object> cell) {
-      this.cell = cell;
+    public SChannelOutput(final SynchronousQueue<Object> cell, final SChannel channel) {
+      this.cell    = cell;
+      this.channel = channel;
     }
 
     public void write(final Object value) throws InterruptedException {
       cell.put(value);
     }
 
+    public void writeAndSuspendReader(final Object value,
+        final boolean doSuspend) throws InterruptedException {
+      channel.breakAfterRead = doSuspend;
+      write(value);
+    }
+
+    public boolean shouldBreakAfterWrite() {
+      return channel.breakAfterWrite;
+    }
+
     @Override
     public SClass getSOMClass() {
       assert ChannelPrimitives.Out != null;

src/som/primitives/actors/CreateActorPrim.java

@@ -13,7 +13,7 @@
 import som.primitives.Primitive;
 import som.primitives.actors.PromisePrims.IsActorModule;
 import som.vm.VmSettings;
-import tools.actors.ActorExecutionTrace;
+import tools.concurrency.ActorExecutionTrace;
 
 
 @GenerateNodeFactory

src/som/primitives/processes/ChannelPrimitives.java

@@ -8,11 +8,14 @@
 import com.oracle.truffle.api.CompilerDirectives.TruffleBoundary;
 import com.oracle.truffle.api.dsl.GenerateNodeFactory;
 import com.oracle.truffle.api.dsl.Specialization;
+import com.oracle.truffle.api.frame.VirtualFrame;
 import com.oracle.truffle.api.source.SourceSection;
 
+import som.VM;
 import som.compiler.AccessModifier;
 import som.compiler.MixinBuilder.MixinDefinitionId;
 import som.interpreter.actors.Actor.UncaughtExceptions;
+import som.interpreter.actors.SuspendExecutionNodeGen;
 import som.interpreter.nodes.nary.BinaryComplexOperation;
 import som.interpreter.nodes.nary.TernaryExpressionNode;
 import som.interpreter.nodes.nary.UnaryExpressionNode;
@@ -33,6 +36,15 @@
 import som.vmobjects.SObject.SImmutableObject;
 import som.vmobjects.SObjectWithClass;
 import som.vmobjects.SSymbol;
+import tools.SourceCoordinate;
+import tools.SourceCoordinate.FullSourceCoordinate;
+import tools.concurrency.Tags.ChannelRead;
+import tools.concurrency.Tags.ChannelWrite;
+import tools.concurrency.Tags.ExpressionBreakpoint;
+import tools.debugger.nodes.AbstractBreakpointNode;
+import tools.debugger.nodes.BreakpointNodeGen;
+import tools.debugger.nodes.DisabledBreakpointNode;
+import tools.debugger.session.Breakpoints;
 
 
 public abstract class ChannelPrimitives {
@@ -58,7 +70,7 @@ public ForkJoinWorkerThread newThread(final ForkJoinPool pool) {
     }
   }
 
-  private static final class ProcessThread extends ForkJoinWorkerThread {
+  public static final class ProcessThread extends ForkJoinWorkerThread {
     ProcessThread(final ForkJoinPool pool) { super(pool); }
   }
 
@@ -119,43 +131,102 @@ public final Object spawn(final SClass procCls, final SArray args) {
     }
   }
 
-  @Primitive(primitive = "procRead:")
+  @Primitive(primitive = "procRead:", selector = "read")
   @GenerateNodeFactory
   public abstract static class ReadPrim extends UnaryExpressionNode {
-    public ReadPrim(final boolean eagerlyWrapped, final SourceSection source) { super(eagerlyWrapped, source); }
+    /** Halt execution when triggered by breakpoint on write end. */
+    @Child protected UnaryExpressionNode haltNode;
+
+    /** Breakpoint info for triggering suspension after write. */
+    @Child protected AbstractBreakpointNode afterWrite;
+
+    public ReadPrim(final boolean eagerlyWrapped, final SourceSection source) {
+      super(eagerlyWrapped, source);
+      haltNode = SuspendExecutionNodeGen.create(false, sourceSection, null);
+
+      if (VmSettings.TRUFFLE_DEBUGGER_ENABLED) {
+        Breakpoints bpCatalog = VM.getWebDebugger().getBreakpoints();
+        FullSourceCoordinate coord = SourceCoordinate.create(source);
+        this.afterWrite = insert(BreakpointNodeGen.create(bpCatalog.getOppositeBreakpoint(coord)));
+      } else {
+        this.afterWrite = insert(new DisabledBreakpointNode());
+      }
+    }
 
     @Specialization
-    public static final Object read(final SChannelInput in) {
+    public final Object read(final VirtualFrame frame, final SChannelInput in) {
       try {
-        return in.read();
+        Object result = in.readAndSuspendWriter(afterWrite.executeCheckIsSetAndEnabled());
+        if (in.shouldBreakAfterRead()) {
+          haltNode.executeEvaluated(frame, result);
+        }
+        return result;
       } catch (InterruptedException e) {
         throw new RuntimeException(e);
       }
     }
+
+    @Override
+    protected boolean isTaggedWithIgnoringEagerness(final Class<?> tag) {
+      if (tag == ChannelRead.class || tag == ExpressionBreakpoint.class) {
+        return true;
+      } else {
+        return super.isTaggedWithIgnoringEagerness(tag);
+      }
+    }
   }
 
-  @Primitive(primitive = "procWrite:val:")
+  @Primitive(primitive = "procWrite:val:", selector = "write:")
   @GenerateNodeFactory
   public abstract static class WritePrim extends BinaryComplexOperation {
     @Child protected IsValue isVal;
 
+    /** Halt execution when triggered by breakpoint on write end. */
+    @Child protected UnaryExpressionNode haltNode;
+
+    /** Breakpoint info for triggering suspension after read. */
+    @Child protected AbstractBreakpointNode afterRead;
+
     public WritePrim(final boolean eagerlyWrapped, final SourceSection source) {
       super(eagerlyWrapped, source);
       isVal = IsValue.createSubNode();
+
+      haltNode = SuspendExecutionNodeGen.create(false, sourceSection, null);
+
+      if (VmSettings.TRUFFLE_DEBUGGER_ENABLED) {
+        Breakpoints bpCatalog = VM.getWebDebugger().getBreakpoints();
+        FullSourceCoordinate coord = SourceCoordinate.create(source);
+        this.afterRead = insert(BreakpointNodeGen.create(bpCatalog.getOppositeBreakpoint(coord)));
+      } else {
+        this.afterRead = insert(new DisabledBreakpointNode());
+      }
     }
 
     @Specialization
-    public final Object write(final SChannelOutput out, final Object val) {
+    public final Object write(final VirtualFrame frame, final SChannelOutput out,
+        final Object val) {
       if (!isVal.executeEvaluated(val)) {
         KernelObj.signalException("signalNotAValueWith:", val);
       }
       try {
-        out.write(val);
+        out.writeAndSuspendReader(val, afterRead.executeCheckIsSetAndEnabled());
+        if (out.shouldBreakAfterWrite()) {
+          haltNode.executeEvaluated(frame, val);
+        }
       } catch (InterruptedException e) {
         throw new RuntimeException(e);
       }
       return val;
     }
+
+    @Override
+    protected boolean isTaggedWithIgnoringEagerness(final Class<?> tag) {
+      if (tag == ChannelWrite.class || tag == ExpressionBreakpoint.class) {
+        return true;
+      } else {
+        return super.isTaggedWithIgnoringEagerness(tag);
+      }
+    }
   }
 
   @Primitive(primitive = "procIn:")

src/som/vmobjects/SSymbol.java

@@ -28,7 +28,7 @@
 
 import som.vm.VmSettings;
 import som.vm.constants.Classes;
-import tools.actors.ActorExecutionTrace;
+import tools.concurrency.ActorExecutionTrace;
 
 public final class SSymbol extends SAbstractObject {
   private final String string;

src/tools/actors/ActorExecutionTrace.java → src/tools/concurrency/ActorExecutionTrace.java

@@ -1,4 +1,4 @@
-package tools.actors;
+package tools.concurrency;
 
 import java.io.BufferedWriter;
 import java.io.File;

src/tools/concurrency/Tags.java

@@ -0,0 +1,26 @@
+package tools.concurrency;
+
+
+public abstract class Tags {
+
+
+  /** Marks the eventual message send operator. */
+  public final class EventualMessageSend extends Tags {
+    private EventualMessageSend() { }
+  }
+
+  /** Marks a read operation on a channel. */
+  public final class ChannelRead extends Tags {
+    private ChannelRead() { }
+  }
+
+  /** Marks a write operation on a channel. */
+  public final class ChannelWrite extends Tags {
+    private ChannelWrite() { }
+  }
+
+  /** Marks an expression that can be target of a breakpoint. */
+  public final class ExpressionBreakpoint extends Tags {
+    private ExpressionBreakpoint() { }
+  }
+}

src/tools/debugger/FrontendConnector.java

@@ -26,8 +26,8 @@
 import som.vmobjects.SSymbol;
 import tools.SourceCoordinate;
 import tools.SourceCoordinate.TaggedSourceCoordinate;
+import tools.concurrency.ActorExecutionTrace;
 import tools.Tagging;
-import tools.actors.ActorExecutionTrace;
 import tools.debugger.frontend.Suspension;
 import tools.debugger.message.Message;
 import tools.debugger.message.Message.OutgoingMessage;
@@ -41,6 +41,7 @@
 import tools.debugger.message.VariablesResponse;
 import tools.debugger.session.AsyncMessageReceiverBreakpoint;
 import tools.debugger.session.Breakpoints;
+import tools.debugger.session.ChannelOppositeBreakpoint;
 import tools.debugger.session.LineBreakpoint;
 import tools.debugger.session.MessageReceiverBreakpoint;
 import tools.debugger.session.MessageSenderBreakpoint;
@@ -285,6 +286,10 @@ public void registerOrUpdate(final PromiseResolverBreakpoint bp) {
     breakpoints.addOrUpdate(bp);
   }
 
+  public void registerOrUpdate(final ChannelOppositeBreakpoint bp) {
+    breakpoints.addOrUpdate(bp);
+  }
+
   public Suspension getSuspension(final int activityId) {
     return webDebugger.getSuspension(activityId);
   }