src/edu/berkeley/fleet/interpreter/InstructionPump.java

   1 package edu.berkeley.fleet.interpreter;
   2 import edu.berkeley.sbp.util.ANSI;
   3 import edu.berkeley.fleet.api.*;
   4 import edu.berkeley.fleet.doc.*;
   5 import edu.berkeley.fleet.api.Instruction;
   6
   7 import java.util.*;
   8
   9 /** anything that has a source (instruction horn) address on the switch fabric */
  10 abstract class InstructionPump extends InterpreterPump {
  11
  12     public int clogged = 0;
  13
  14     /** the currently executing instruction */
  15     private Instruction.Executable executing = null;
  16
  17     /** all instructions waiting to be executed (excludes executing) */
  18     private Queue<Instruction.Executable> instructions = new LinkedList<Instruction.Executable>();
  19
  20     /** count of how many "standing instruction only" kills remain to be executed */
  21     private int killNextStandingInstruction = 0;
  22
  23     InstructionPump(InterpreterShip ship, String name, String[] ports) {
  24         super(ship, name, ports);
  25     }
  26
  27     public void massacre() {
  28         executing = null;
  29         while(instructions.size() > 0)
  30             instructions.remove();
  31     }
  32
  33     public void kill(int count, boolean killOnlyStandingInstructions) {
  34         if (count==0) return;
  35         if (!killOnlyStandingInstructions) {
  36             if (executing != null) { executing = null; count--; }
  37             for(;count > 0;count--) {
  38                 if (instructions.size()==0)
  39                     throw new RuntimeException("deadlocked " + this + " by killing too many instructions");
  40                 instructions.remove();
  41             }
  42         } else {
  43             if (count != 1) throw new RuntimeException("this should never happen");
  44             this.killNextStandingInstruction++;
  45         }
  46     }
  47
  48     /** an instruction arrives from the instruction horn */
  49     void addInstruction(Instruction.Executable instr) {
  50         if (killNextStandingInstruction > 0) { /* FIXME technically we should refuse to take the next instruction here */ }
  51         instructions.add(instr);
  52     }
  53
  54     protected void shutdown(boolean leaveAsInbox) {
  55         if (!(executing != null || instructions.size() > 0)) return;
  56         Log.println(ANSI.red(" WARNING: you left instructions on the instruction queue of port " +
  57                              this + "; they are:"));
  58         if (executing != null)
  59             Log.println("   " + executing);
  60         for(Instruction.Executable i : instructions)
  61             Log.println("   " + i);
  62     }
  63
  64     // interface to subclass ///////////////////////////////////////////////////////////////////////
  65
  66     /** this will be invoked periodically; should return true to "consume" an instruction, false to leave it executing */
  67     protected abstract boolean service(Instruction.Executable instr);
  68
  69     protected final void service() {
  70         if (executing == null)
  71             if (instructions.size() > 0)
  72                 executing = instructions.remove();
  73             else
  74                 return;
  75
  76         if (executing != null && killNextStandingInstruction>0 && executing.isStanding()) {
  77             executing = null;
  78             killNextStandingInstruction--;
  79             return;
  80         }
  81
  82         boolean ret = service(executing);
  83         if (!ret) return;
  84         if (executing.isRequeueing()) {
  85             executing = executing.decrementCount();
  86             if (executing != null)
  87                 addInstruction(executing);
  88             executing = null;
  89
  90         } else {
  91             if (executing != null && !executing.isStanding())
  92                 executing = executing.decrementCount();
  93         }
  94     }
  95
  96
  97 }