computation requests on the tenor side

src/tenor/java/tenor/Bot.java

@@ -17,19 +17,24 @@
 
 package tenor;
 
+import cabaletta.comms.BufferedConnection;
 import cabaletta.comms.IConnection;
 import cabaletta.comms.IMessageListener;
+import cabaletta.comms.iMessage;
+import cabaletta.comms.upward.MessageComputationResponse;
 import cabaletta.comms.upward.MessageStatus;
 
+import java.io.IOException;
+
 public class Bot implements IMessageListener {
     public final BotTaskRegistry taskRegistry = new BotTaskRegistry(this);
 
-    private final IConnection connectionToBot;
+    private final BufferedConnection connectionToBot;
     private volatile MessageStatus mostRecentUpdate;
 
     public Bot(IConnection conn) {
-        this.connectionToBot = conn;
+        this.connectionToBot = BufferedConnection.makeBuffered(conn);
-        // TODO event loop to read messages non blockingly
+        // TODO event loop calling tick?
     }
 
     public int getCurrentQuantityInInventory(String item) {
@@ -37,8 +42,72 @@ public class Bot implements IMessageListener {
         throw new UnsupportedOperationException("oppa");
     }
 
+    public void tick() {
+        // TODO i have no idea what tenor's threading model will be
+        // probably single threaded idk
+        ComputationRequestManager.INSTANCE.dispatchAll();
+        receiveMessages();
+        TaskLeaf<?> curr = decideWhatToDoNow();
+        iMessage command = doTheTask(curr);
+        try {
+            connectionToBot.sendMessage(command);
+        } catch (IOException e) {
+            e.printStackTrace();
+            disconnect();
+        }
+    }
+
+    public void receiveMessages() {
+        try {
+            connectionToBot.handleAllPendingMessages(this);
+        } catch (IOException e) {
+            e.printStackTrace();
+            disconnect();
+        }
+    }
+
+    public void disconnect() {
+        // TODO i have no idea how to handle this
+        // this destroys the task tree
+        // wew lad?
+        connectionToBot.close();
+    }
+
+    public TaskLeaf<?> decideWhatToDoNow() {
+        // TODO idk lol
+        throw new UnsupportedOperationException();
+    }
+
+    public iMessage doTheTask(TaskLeaf<?> task) {
+        // TODO idk lol
+        throw new UnsupportedOperationException();
+    }
+
+    public void send(iMessage msg) {
+        try {
+            connectionToBot.sendMessage(msg);
+        } catch (IOException e) {
+            e.printStackTrace();
+            disconnect();
+        }
+    }
+
+    public MessageStatus getMostRecentUpdate() {
+        return mostRecentUpdate;
+    }
+
+    public String getHostIdentifier() {
+        // return ((SocketConnection) ((BufferedConnection) connectionToBot).getUnderlying()).getSocket().getHostname()
+        return "localhost";
+    }
+
     @Override
     public void handle(MessageStatus msg) {
         mostRecentUpdate = msg;
     }
+
+    @Override
+    public void handle(MessageComputationResponse msg) {
+        ComputationRequestManager.INSTANCE.onRecv(this, msg);
+    }
 }

src/tenor/java/tenor/ComputationRequest.java

@@ -0,0 +1,93 @@
+/*
+ * This file is part of Baritone.
+ *
+ * Baritone is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Lesser General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * Baritone is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public License
+ * along with Baritone.  If not, see <https://www.gnu.org/licenses/>.
+ */
+
+package tenor;
+
+import cabaletta.comms.upward.MessageComputationResponse;
+
+import java.util.HashSet;
+import java.util.Set;
+
+public class ComputationRequest {
+    public final Bot bot;
+    public final String goal;
+    private final Set<SingularTaskLeaf> parents; // TODO quantized UGH
+    private MessageComputationResponse resp;
+
+
+    ComputationRequest(Bot bot, String goal) {
+        this.bot = bot;
+        this.goal = goal;
+        this.parents = new HashSet<>();
+        this.resp = null;
+    }
+
+    public MessageComputationResponse getResp() {
+        return resp;
+    }
+
+    public void receivedResponse(MessageComputationResponse mcrn) {
+        resp = mcrn;
+        // TODO notify parent tasks?
+    }
+
+    public Status getStatus() {
+        // :woke
+        // make this a huge nested ternary when
+        if (resp != null) {
+            if (resp.pathLength == 0) {
+                return Status.RECV_FAILURE;
+            } else {
+                return Status.RECV_SUCCESS;
+            }
+        } else {
+            if (ComputationRequestManager.INSTANCE.inProgress(this)) {
+                return Status.IN_PROGRESS;
+            } else {
+                return Status.QUEUED;
+            }
+        }
+    }
+
+    public void addParentTask(SingularTaskLeaf parent) {
+        parents.add(parent);
+    }
+
+    // hmmmmm this has parent tasks and has a cost and a priority.... god forbid should i make this a SingularTaskLeaf itself?!?!??!?
+    public Double cost() {
+        switch (getStatus()) {
+            case RECV_FAILURE:
+                return 1000000000D;
+            case RECV_SUCCESS:
+                return resp.pathCost;
+            case IN_PROGRESS:
+            case QUEUED:
+                return null;
+        }
+    }
+
+    public double priority() {
+        return parents.stream().mapToDouble(SingularTaskLeaf::priority).sum();
+    }
+
+    public enum Status {
+        QUEUED,
+        IN_PROGRESS,
+        RECV_SUCCESS,
+        RECV_FAILURE,
+    }
+}

src/tenor/java/tenor/ComputationRequestManager.java

@@ -0,0 +1,74 @@
+/*
+ * This file is part of Baritone.
+ *
+ * Baritone is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Lesser General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * Baritone is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public License
+ * along with Baritone.  If not, see <https://www.gnu.org/licenses/>.
+ */
+
+package tenor;
+
+import cabaletta.comms.downward.MessageComputationRequest;
+import cabaletta.comms.upward.MessageComputationResponse;
+import cabaletta.comms.upward.MessageStatus;
+
+import java.security.SecureRandom;
+import java.util.*;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+public enum ComputationRequestManager {
+    INSTANCE;
+    public static final int MAX_SIMUL_COMPUTATIONS_PER_HOST = 2;
+
+    private Map<Long, ComputationRequest> inProgress = new HashMap<>();
+    private Map<Bot, Map<String, ComputationRequest>> byGoal = new HashMap<>();
+    public Function<Bot, String> botToHostMapping = Bot::toString; // TODO map bots to hosts by connection maybe?
+
+
+    public ComputationRequest getByGoal(SingularTaskLeaf task, String goal) {
+        Map<String, ComputationRequest> thisBot = byGoal.computeIfAbsent(task.bot, x -> new HashMap<>());
+        ComputationRequest req = thisBot.computeIfAbsent(goal, g -> new ComputationRequest(task.bot, goal));
+        req.addParentTask(task);
+        return req;
+    }
+
+    public void dispatchAll() {
+        Map<String, List<ComputationRequest>> highestPriorityByHost = byGoal.values().stream().map(Map::values).flatMap(Collection::stream).filter(c -> c.getStatus() == ComputationRequest.Status.QUEUED).collect(Collectors.groupingBy(req -> botToHostMapping.apply(req.bot), Collectors.collectingAndThen(Collectors.toList(), l -> l.stream().sorted(Comparator.comparingDouble(ComputationRequest::priority).reversed()).collect(Collectors.toList()))));
+        inProgress.values().stream().collect(Collectors.groupingBy(req -> botToHostMapping.apply(req.bot), Collectors.collectingAndThen(Collectors.counting(), count -> MAX_SIMUL_COMPUTATIONS_PER_HOST - count))).entrySet().stream().flatMap(entry -> highestPriorityByHost.get(entry.getKey()).subList(0, entry.getValue().intValue()).stream()).forEach(this::dispatch);
+    }
+
+    private void dispatch(ComputationRequest req) {
+        if (inProgress(req)) {
+            throw new IllegalStateException();
+        }
+        long key = new SecureRandom().nextLong();
+        inProgress.put(key, req);
+        MessageStatus status = req.bot.getMostRecentUpdate();
+        req.bot.send(new MessageComputationRequest(key, status.pathStartX, status.pathStartY, status.pathStartZ, req.goal));
+    }
+
+    public void onRecv(Bot receivedFrom, MessageComputationResponse mcrn) {
+        ComputationRequest req = inProgress.remove(mcrn.computationID);
+        if (req == null) {
+            throw new IllegalStateException("Received completed computation that we didn't ask for");
+        }
+        if (!Objects.equals(req.bot, receivedFrom)) {
+            throw new IllegalStateException("Received completed computation from the wrong connection?????");
+        }
+        req.receivedResponse(mcrn);
+    }
+
+    public boolean inProgress(ComputationRequest req) {
+        return inProgress.values().contains(req);
+    }
+}