feat: add configurable row/column topic map

simulator/config/gossipdas1k.cfg

@@ -106,7 +106,7 @@ init.1uniqueNodeID.protocolEvildas 8dasprotocol
 init.1uniqueNodeID.validator_rate 1.0
 init.1uniqueNodeID.evilNodeRatioValidator 0.0
 init.1uniqueNodeID.evilNodeRatioNonValidator 0.0
-
+init.1uniqueNodeID.colrow_topic 1
 
 #Adds initial state to the routing tables
 #init.2statebuilder peersim.kademlia.StateBuilder

simulator/src/main/java/peersim/kademlia/das/CustomDistributionGossipDas.java

@@ -31,7 +31,7 @@ public class CustomDistributionGossipDas implements peersim.core.Control {
   private static final String PAR_EVIL_RATIO_NONVAL = "evilNodeRatioNonValidator";
 
   private static final String PAR_VALIDATOR_RATE = "validator_rate";
-
+  private static final String PAR_ROWCOL_TOPIC = "colrow_topic";
   /** Protocol identifiers for Kademlia, DAS, etc. * */
   private int protocolKadID;
 
@@ -51,6 +51,8 @@ public class CustomDistributionGossipDas implements peersim.core.Control {
   private BigInteger builderAddress;
   private UniformRandomGenerator urg;
 
+  private GossipTopicMap topicMap;
+
   public CustomDistributionGossipDas(String prefix) {
     protocolKadID = Configuration.getPid(prefix + "." + PAR_PROT_KAD);
     protocolDasBuilderID = Configuration.getPid(prefix + "." + PAR_PROT_DAS_BUILDER);
@@ -63,6 +65,8 @@ public CustomDistributionGossipDas(String prefix) {
     evilRatioNonValidator = Configuration.getDouble(prefix + "." + PAR_EVIL_RATIO_NONVAL, 0.0);
     urg = new UniformRandomGenerator(KademliaCommonConfig.BITS, CommonState.r);
     validatorRate = Configuration.getDouble(prefix + "." + PAR_VALIDATOR_RATE, 1.0);
+    int numRowsColsTOpic = Configuration.getInt(prefix + "." + PAR_ROWCOL_TOPIC, 1);
+    topicMap = new GossipTopicMap(numRowsColsTOpic);
   }
 
   public boolean execute() {
@@ -122,7 +126,7 @@ public boolean execute() {
         nonValidatorsIds.add(gossipProt.getGossipNode().getId());
       }
 
-      dasProt.setGossipProtocol(generalNode, gossipProt);
+      dasProt.setGossipProtocol(generalNode, gossipProt, topicMap);
       dasProt.setProtocolId(protocolDasBuilderID);
       // gossipProt.setEventsCallback(dasProt);
 

simulator/src/main/java/peersim/kademlia/das/GossipDAS.java

@@ -50,6 +50,7 @@ public abstract class GossipDAS implements Cloneable, EDProtocol, GossipEvent {
   private int tid;
   protected int protocolId;
   protected boolean isBuilder, isValidator;
+  protected GossipTopicMap topicMap;
 
   public GossipDAS(String prefix) {
     GossipDAS.prefix = prefix;
@@ -144,8 +145,9 @@ protected void handleInitNewBlock(Message m, int myPid) {
    *
    * @param prot GossipSubProtocol
    */
-  public void setGossipProtocol(Node node, GossipSubProtocol prot) {
+  public void setGossipProtocol(Node node, GossipSubProtocol prot, GossipTopicMap topicMap) {
     this.gossipsub = prot;
+    this.topicMap = topicMap;
     transport = (BwTransport) (Network.prototype).getProtocol(tid);
     transport.setBw(node, bw);
     this.gossipsub.setTransport(this.transport);

simulator/src/main/java/peersim/kademlia/das/GossipDASBuilder.java

@@ -28,25 +28,33 @@ protected void handleInitNewBlock(Message m, int myPid) {
     if (!started) {
       started = true;
 
+      String rowTopic = "", columnTopic = "";
       for (int j = 1; j <= KademliaCommonConfigDas.BLOCK_DIM_SIZE; j++) {
-        String topic = "Row" + j;
-        gossipsub.Join(topic);
-        GossipSubProtocol.getTable().addPeer(topic, gossipsub.getGossipNode().getId());
-        topic = "Column" + j;
-        gossipsub.Join(topic);
-        GossipSubProtocol.getTable().addPeer(topic, gossipsub.getGossipNode().getId());
+        String topic = topicMap.getRowTopic(j);
+        if (rowTopic != topic) {
+          gossipsub.Join(topic);
+          GossipSubProtocol.getTable().addPeer(topic, gossipsub.getGossipNode().getId());
+        }
+        rowTopic = topic;
+
+        topic = topicMap.getColumnTopic(j);
+        if (columnTopic != topic) {
+          gossipsub.Join(topic);
+          GossipSubProtocol.getTable().addPeer(topic, gossipsub.getGossipNode().getId());
+        }
+        columnTopic = topic;
       }
     } else {
       for (int i = 1; i <= KademliaCommonConfigDas.BLOCK_DIM_SIZE; i++) {
         Sample[] samples = currentBlock.getSamplesByRow(i);
-        String topic = "Row" + i;
+        String topic = topicMap.getRowTopic(i);
         Message msg =
             Message.makePublishMessage(topic, Arrays.copyOfRange(samples, 0, samples.length / 2));
         msg.src = this.gossipsub.node;
         gossipsub.Publish(msg, myPid);
 
         samples = currentBlock.getSamplesByColumn(i);
-        topic = "Column" + i;
+        topic = topicMap.getColumnTopic(i);
         msg = Message.makePublishMessage(topic, Arrays.copyOfRange(samples, 0, samples.length / 2));
         msg.src = this.gossipsub.node;
         gossipsub.Publish(msg, myPid);

simulator/src/main/java/peersim/kademlia/das/GossipDASValidator.java

@@ -4,7 +4,9 @@
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.HashMap;
+import java.util.HashSet;
 import java.util.List;
+import java.util.Set;
 import peersim.config.Configuration;
 import peersim.core.CommonState;
 import peersim.kademlia.KademliaObserver;
@@ -19,6 +21,8 @@ public class GossipDASValidator extends GossipDAS {
   protected boolean started;
   int row1, column1, row2, column2;
   protected HashMap<BigInteger, List<Message>> missingSamples;
+  protected HashMap<String, List<Long>> operationTopicMap;
+  Set<String> topics;
 
   public GossipDASValidator(String prefix) {
     super(prefix);
@@ -28,6 +32,8 @@ public GossipDASValidator(String prefix) {
     row1 = column1 = row2 = column2 = 0;
     missingSamples = new HashMap<>();
     bw = Configuration.getInt(prefix + "." + PAR_BW, KademliaCommonConfigDas.VALIDATOR_UPLOAD_RATE);
+    operationTopicMap = new HashMap<>();
+    topics = new HashSet<>();
   }
 
   @Override
@@ -41,27 +47,20 @@ protected void handleInitNewBlock(Message m, int myPid) {
     logger.warning("Validator Init block");
 
     if (!started) {
-      String topic;
       started = true;
+
       row1 = CommonState.r.nextInt(KademliaCommonConfigDas.BLOCK_DIM_SIZE) + 1;
-      topic = "Row" + row1;
-      gossipsub.Join(topic);
-      GossipSubProtocol.getTable().addPeer(topic, gossipsub.getGossipNode().getId());
+      JoinTopic(topicMap.getRowTopic(row1), row1);
 
       column1 = CommonState.r.nextInt(KademliaCommonConfigDas.BLOCK_DIM_SIZE) + 1;
-      topic = "Column" + column1;
-      gossipsub.Join(topic);
-      GossipSubProtocol.getTable().addPeer(topic, gossipsub.getGossipNode().getId());
+      JoinTopic(topicMap.getColumnTopic(column1), column1);
 
       row2 = CommonState.r.nextInt(KademliaCommonConfigDas.BLOCK_DIM_SIZE) + 1;
-      topic = "Row" + row2;
-      gossipsub.Join(topic);
-      GossipSubProtocol.getTable().addPeer(topic, gossipsub.getGossipNode().getId());
+      JoinTopic(topicMap.getRowTopic(row2), row2);
 
       column2 = CommonState.r.nextInt(KademliaCommonConfigDas.BLOCK_DIM_SIZE) + 1;
-      topic = "Column" + column2;
-      gossipsub.Join(topic);
-      GossipSubProtocol.getTable().addPeer(topic, gossipsub.getGossipNode().getId());
+      JoinTopic(topicMap.getRowTopic(column2), column2);
+
     } else {
       createValidatorSamplingOperation(row1, 0, CommonState.getTime(), null);
       createValidatorSamplingOperation(0, column1, CommonState.getTime(), null);
@@ -125,21 +124,26 @@ public void messageReceived(Message m) {
     }
     toSend.clear();
 
-    long id;
+    /*long id;
+    System.out.println(topic);
     if (topic.contains("Column")) {
       id = Long.parseLong(topic.replace("Column", ""));
     } else {
       id = Long.parseLong(topic.replace("Row", ""));
-    }
+    }*/
 
     // logger.info("Sample received row:" + s.getRow() + " column:" + s.getColumn());
-    if (samplingOp.get(id) != null) {
-      SamplingOperation op = samplingOp.get(id);
-      op.elaborateResponse(samples);
-      logger.warning("Operation found:" + op.getSamples().length);
-      if (op.completed()) {
-        KademliaObserver.reportOperation(op);
-        logger.warning("Sampling operation completed " + op.getId());
+    List<Long> ops = operationTopicMap.get(topic);
+
+    for (Long id : ops) {
+      if (samplingOp.get(id) != null) {
+        SamplingOperation op = samplingOp.get(id);
+        op.elaborateResponse(samples);
+        logger.warning("Operation found:" + op.getSamples().length);
+        if (op.completed()) {
+          KademliaObserver.reportOperation(op);
+          logger.warning("Sampling operation completed " + op.getId());
+        }
       }
     }
   }
@@ -304,4 +308,18 @@ protected Message generateGetSampleMessage(BigInteger[] sampleId) {
 
     return m;
   }
+
+  private void JoinTopic(String topic, int value) {
+
+    if (!topics.contains(topic)) {
+      gossipsub.Join(topic);
+      GossipSubProtocol.getTable().addPeer(topic, gossipsub.getGossipNode().getId());
+      topics.add(topic);
+      List<Long> ops = new ArrayList<>();
+      ops.add((long) value);
+      operationTopicMap.put(topic, ops);
+    } else {
+      operationTopicMap.get(topic).add((long) value);
+    }
+  }
 }

simulator/src/main/java/peersim/kademlia/das/GossipTopicMap.java

@@ -0,0 +1,29 @@
+package peersim.kademlia.das;
+
+public class GossipTopicMap {
+  private static final int SIZE = 512;
+  private final int groupSize;
+
+  public GossipTopicMap(int groupSize) {
+    if (groupSize <= 0 || groupSize > SIZE) {
+      throw new IllegalArgumentException("Group size must be between 1 and " + SIZE);
+    }
+    this.groupSize = groupSize;
+  }
+
+  public String getRowTopic(int row) {
+    validateIndex(row);
+    return "Row" + ((row / groupSize) + 1);
+  }
+
+  public String getColumnTopic(int column) {
+    validateIndex(column);
+    return "Column" + ((column / groupSize) + 1);
+  }
+
+  private void validateIndex(int index) {
+    if (index < 1 || index > SIZE) {
+      throw new IllegalArgumentException("Index must be between 0 and " + (SIZE - 1));
+    }
+  }
+}