Актеры Java с Килимом

Я наконец-то нашел время для переноса своего кода вызова актерского процесса ( Erlang , Scala ) для работы с библиотекой Kilim сегодня вечером.

Kilim — это библиотека Java для облегченной передачи сообщений. Три основные вещи, о которых нужно знать: Задачи, Почтовые ящики и @pausable. В Kilim актеры заменены Задачами, которые являются просто облегченными управляемыми объектами, которые выполняют в основном ту же роль.

Почтовые ящики предназначены для доступа одного производителя к нескольким производителям так же, как почтовые ящики Erlang и Scala. В Kilim, тем не менее, Mailbox — это просто класс, и задачи не обязательно имеют почтовый ящик или даже могут иметь более одного. Почтовые ящики также являются общими и набираются по типу сообщений, которые они должны получать. Это открывает новые способы объединения задач и почтовых ящиков в более широкий спектр структур. Сообщения, как правило, простые (потенциально изменяемые) классы.

Наконец, есть аннотация под названием @pausable. Это используется для указания того, что метод может быть приостановлен в стиле продолжения во время приостановленных вызовов; sleep и yield являются двумя предоставляемыми хуками, а методы get / put почтового ящика также могут быть приостановлены. @pausable также используется для обозначения классов для инструментовки.

Это приводит меня к ткачу во время компиляции. После того, как вы скомпилировали свои классы, вам нужно запустить ткач времени компиляции, чтобы изменить байт-код так, чтобы была доступна приостановка в стиле продолжения. Во время выполнения пауза используется для планирования большого количества участников по небольшому количеству потоков ядра.

В целом, это действительно интересный набор функций, который очень хорошо адаптирует модель актера Эрланга / Scala в статически типизированный мир Java.

С точки зрения использования, это отчасти боль. Насколько я могу судить, ткачество во время компиляции — это отстой, потому что в каждой цепочке инструментов возникает излом, плюс не происходит ничего, что нельзя было бы сделать во время выполнения с помощью агента Java. У меня также было много времени, чтобы заставить Уивера выполнить мой первый удар по коду. Было совершенно не очевидно, что сообщения об ошибках указывают на то, что я забыл @pausable в классах не-Actor. К счастью, я знаю достаточно о ASM, чтобы сказать, чего мне не хватало. И как только я запустил его, у меня появились странные ошибки NoSuchMethodErrors из-за неправильного указания @pausable для методов, которые в этом не нуждались.

Эти неровности меня не слишком беспокоили — это новый проект, и я понимаю, что еще рано для такой помощи.

Теперь немного кода. Это действительно довольно большой порт из кода Scala в Java, который был довольно близок. Я разбил код (ранее в одном файле) на Ring (основной код), Message, TokenMessage, NodeActor и TimerActor. Вы заметите, что в Java-версии гораздо больше кода, чем в Scala или Erlang.

import java.util.ArrayList;   
import java.util.List;   
  
import kilim.Mailbox;   
import kilim.pausable;   
  
@pausable  
public class Ring {   
  
  public static void main(String arg[]) {   
    new Ring().startRing(Integer.parseInt(arg[0]));   
  }   
     
  public void startRing(int n) {   
    List<NodeActor> nodes = spawnNodes(n, startTimer());   
    Mailbox<Message> mailbox = connectNodes(n, nodes);   
    mailbox.putnb(Message.START);   
       
    try { Thread.sleep(100000000); } catch(InterruptedException e) {}   
  }   
     
  private TimerActor startTimer() {   
    TimerActor actor = new TimerActor(new Mailbox<Message>());   
    actor.start();   
    return actor;   
  }   
     
  private List<NodeActor> spawnNodes(int n, TimerActor timer) {   
    System.out.println("constructing nodes");   
    long startConstructing = System.currentTimeMillis();   
       
    List<NodeActor> nodes = new ArrayList<NodeActor>(n+1);   
    for(int i=0; i<n; i++) {   
      nodes.add(new NodeActor(i, new Mailbox<Message>(), timer.getInbox()));   
      nodes.get(i).start();   
    }       
  
    long endConstructing = System.currentTimeMillis();   
    System.out.println("Took " + (endConstructing-startConstructing) + " ms to construct " + n + " nodes");   
    return nodes;   
  }   
     
  private Mailbox<Message> connectNodes(int n, List<NodeActor> nodes) {   
    System.out.println("connecting nodes");   
    nodes.add(nodes.get(0));   
    for(int i=0; i<n; i++) {   
      nodes.get(i).connect(nodes.get(i+1).getInbox());   
    }   
    return nodes.get(0).getInbox();   
  }     
}   

Очень важно, чтобы класс Ring был помечен как @pausable, иначе ткач не будет работать. Теперь классы сообщений. Я определил несколько неизменяемых одноэлементных сообщений в Message и изменяемый TokenMessage:

public class Message {   
  public static final Message START = new Message();   
  public static final Message STOP = new Message();   
  public static final Message CANCEL = new Message();   
}   
  
public class TokenMessage extends Message {   
  public final int source;   
  public int value;   
     
  public TokenMessage(int source, int value) {   
    this.source = source;   
    this.value = value;   
  }   
} 

And here’s the node actor translated into a Kilim Task (note the @pausable annotation on the execute() method, which is defined in Task):

import kilim.Mailbox;   
import kilim.Task;   
import kilim.pausable;   
  
public class NodeActor extends Task {   
  private final int nodeId;   
  private final Mailbox<Message> inbox;   
  private final Mailbox<Message> timerInbox;   
  private Mailbox<Message> nextInbox;   
     
  public NodeActor(int nodeId, Mailbox<Message> inbox, Mailbox<Message> timerInbox) {   
    this.nodeId = nodeId;   
    this.inbox = inbox;   
    this.timerInbox = timerInbox;   
  }   
     
  public Mailbox<Message> getInbox() {    
    return this.inbox;   
  }   
     
  public void connect(Mailbox<Message> nextInbox) {   
    this.nextInbox = nextInbox;   
  }   
     
  @pausable  
  public void execute() throws Exception {   
    while(true) {   
      Message message = inbox.get();   
      if(message.equals(Message.START)) {   
        System.out.println(System.currentTimeMillis() + " " + nodeId + ": Starting messages");   
        timerInbox.putnb(Message.START);   
        nextInbox.putnb(new TokenMessage(nodeId, 0));   
      } else if(message.equals(Message.STOP)) {   
        //System.out.println(System.currentTimeMillis() + " " + nodeId + ": Stopping");   
        nextInbox.putnb(Message.STOP);   
        break;   
      } else if(message instanceof TokenMessage) {   
        TokenMessage token = (TokenMessage)message;   
        if(token.source == nodeId) {   
          int nextVal = token.value+1;   
          if(nextVal % 10000 == 0) {   
            System.out.println(System.currentTimeMillis() + " " + nodeId + ": Around ring " + nextVal + " times");   
          }   
             
          if(nextVal == 1000000) {   
            timerInbox.putnb(Message.STOP);   
            timerInbox.putnb(Message.CANCEL);   
            nextInbox.putnb(Message.STOP);   
            break;   
          } else {   
            token.value = nextVal;   
            nextInbox.putnb(token);   
          }   
        } else {   
          nextInbox.putnb(token);   
        }   
      }   
    }   
  }   
}   

And for completeness, here’s the Timer Actor:

import kilim.Mailbox;   
import kilim.Task;   
import kilim.pausable;   
  
public class TimerActor extends Task {   
  private final Mailbox<Message> inbox;   
  private boolean timing;   
  private long startTime;   
     
  public TimerActor(Mailbox<Message> inbox) {   
    this.inbox = inbox;   
  }   
     
  public Mailbox<Message> getInbox() {    
    return this.inbox;   
  }   
     
  @pausable  
  public void execute() throws Exception {   
    while(true) {   
      Message message = inbox.get();   
      if(message.equals(Message.START) && !timing) {   
        startTime = System.currentTimeMillis();   
        timing = true;   
      } else if(message.equals(Message.STOP) && timing) {   
        long endTime = System.currentTimeMillis();   
        System.out.println("Start=" + startTime + " Stop=" + endTime + " Elapsed=" + (endTime-startTime));   
        timing = false;   
      } else if(message.equals(Message.CANCEL)) {   
        break;   
      }   
    }   
  }   
}  

To compile and weave, you’ll do something like this:

$ export KCP=$KILIM/libs/asm-all-2.2.3.jar:$KILIM/classes  

$ javac -cp $KCP -d bin src/*.java  

$ java -cp $KCP kilim.tools.Weaver -d ./bin ./bin  

$ java -cp $KCPL./bin Ring 100 

And finally the results, here in comparison with Erlang and Scala 2.7.3/JDK 1.6:

So, Kilim demonstrates process creation faster than Scala (but still slower than Erlang) and message-passing in the realm of Erlang and significantly faster than Scala. That’s pretty darn impressive! I’m looking forward to watching where this library goes.

From http://tech.puredanger.com/