Talk about GossipProtocol of scalecube-cluster

  gossip

Order

This paper mainly studies GossipProtocol of scalecube-cluster

GossipProtocol

scalecube-cluster-2.2.5/cluster/src/main/java/io/scalecube/cluster/gossip/GossipProtocol.java

/**
 * Gossip Protocol component responsible for spreading information (gossips) over the cluster
 * members using infection-style dissemination algorithms. It provides reliable cross-cluster
 * broadcast.
 */
public interface GossipProtocol {

  /** Starts running gossip protocol. After started it begins to receive and send gossip messages */
  void start();

  /** Stops running gossip protocol and releases occupied resources. */
  void stop();

  /**
   * Spreads given message between cluster members.
   *
   * @return future result with gossip id once gossip fully spread.
   */
  Mono<String> spread(Message message);

  /** Listens for gossips from other cluster members. */
  Flux<Message> listen();
}
  • GossipProtocol interface defines start, stop, spread, listen methods.

GossipProtocolImpl

scalecube-cluster-2.2.5/cluster/src/main/java/io/scalecube/cluster/gossip/GossipProtocolImpl.java

public final class GossipProtocolImpl implements GossipProtocol {

  private static final Logger LOGGER = LoggerFactory.getLogger(GossipProtocolImpl.class);

  // Qualifiers

  public static final String GOSSIP_REQ = "sc/gossip/req";

  // Injected

  private final Member localMember;
  private final Transport transport;
  private final GossipConfig config;

  // Local State

  private long currentPeriod = 0;
  private long gossipCounter = 0;
  private Map<String, GossipState> gossips = new HashMap<>();
  private Map<String, MonoSink<String>> futures = new HashMap<>();

  private List<Member> remoteMembers = new ArrayList<>();
  private int remoteMembersIndex = -1;

  // Disposables

  private final Disposable.Composite actionsDisposables = Disposables.composite();

  // Subject

  private final FluxProcessor<Message, Message> subject =
      DirectProcessor.<Message>create().serialize();

  private final FluxSink<Message> sink = subject.sink();

  // Scheduled

  private final Scheduler scheduler;

  /**
   * Creates new instance of gossip protocol with given memberId, transport and settings.
   *
   * @param localMember local cluster member
   * @param transport cluster transport
   * @param membershipProcessor membership event processor
   * @param config gossip protocol settings
   * @param scheduler scheduler
   */
  public GossipProtocolImpl(
      Member localMember,
      Transport transport,
      Flux<MembershipEvent> membershipProcessor,
      GossipConfig config,
      Scheduler scheduler) {

    this.transport = Objects.requireNonNull(transport);
    this.config = Objects.requireNonNull(config);
    this.localMember = Objects.requireNonNull(localMember);
    this.scheduler = Objects.requireNonNull(scheduler);

    // Subscribe
    actionsDisposables.addAll(
        Arrays.asList(
            membershipProcessor //
                .publishOn(scheduler)
                .subscribe(this::onMemberEvent, this::onError),
            transport
                .listen()
                .publishOn(scheduler)
                .filter(this::isGossipReq)
                .subscribe(this::onGossipReq, this::onError)));
  }

  @Override
  public void start() {
    actionsDisposables.add(
        scheduler.schedulePeriodically(
            this::doSpreadGossip,
            config.getGossipInterval(),
            config.getGossipInterval(),
            TimeUnit.MILLISECONDS));
  }

  @Override
  public void stop() {
    // Stop accepting gossip requests and spreading gossips
    actionsDisposables.dispose();

    // Stop publishing events
    sink.complete();
  }

  @Override
  public Mono<String> spread(Message message) {
    return Mono.fromCallable(() -> message)
        .subscribeOn(scheduler)
        .flatMap(msg -> Mono.create(sink -> futures.put(createAndPutGossip(msg), sink)));
  }

  @Override
  public Flux<Message> listen() {
    return subject.onBackpressureBuffer();
  }

  private void onMemberEvent(MembershipEvent event) {
    Member member = event.member();
    if (event.isRemoved()) {
      remoteMembers.remove(member);
    }
    if (event.isAdded()) {
      remoteMembers.add(member);
    }
  }

  private void onGossipReq(Message message) {
    long period = this.currentPeriod;
    GossipRequest gossipRequest = message.data();
    for (Gossip gossip : gossipRequest.gossips()) {
      GossipState gossipState = gossips.get(gossip.gossipId());
      if (gossipState == null) { // new gossip
        gossipState = new GossipState(gossip, period);
        gossips.put(gossip.gossipId(), gossipState);
        sink.next(gossip.message());
      }
      gossipState.addToInfected(gossipRequest.from());
    }
  }

  private boolean isGossipReq(Message message) {
    return GOSSIP_REQ.equals(message.qualifier());
  }

  private String createAndPutGossip(Message message) {
    long period = this.currentPeriod;
    Gossip gossip = new Gossip(generateGossipId(), message);
    GossipState gossipState = new GossipState(gossip, period);
    gossips.put(gossip.gossipId(), gossipState);
    return gossip.gossipId();
  }

  //......

}
  • GossipProtocol interface is implemented by Gossip Protocol, which maintains the gossipId named gossips, the map of GossipState, and the list of remoteMembers.
  • Its constructor subscribes to the member; hipProcessor and triggers the onMemberEvent method, which adds or removes members to remoteMembers according to the MembershipEvent; Subscribed to transport.listen (), filtered out gossipReq and triggered ongossipReq method. The method merges gossips of GossipRequest into local gossips, and sends message for new Gossip to sink, maintains gossipState of Gossip, and adds the requested memberId to infected; The spread method places the message into the local gossips
  • The start method executes the doSpreadGossip method every gossipInterval; The spread method creates Gossip through createAndPutGossip and puts it into gossips

doSpreadGossip

scalecube-cluster-2.2.5/cluster/src/main/java/io/scalecube/cluster/gossip/GossipProtocolImpl.java

public final class GossipProtocolImpl implements GossipProtocol {

  //......

  private List<Member> remoteMembers = new ArrayList<>();

  private int remoteMembersIndex = -1;

  private void doSpreadGossip() {
    // Increment period
    long period = currentPeriod++;

    // Check any gossips exists
    if (gossips.isEmpty()) {
      return; // nothing to spread
    }

    try {
      // Spread gossips to randomly selected member(s)
      selectGossipMembers().forEach(member -> spreadGossipsTo(period, member));

      // Sweep gossips
      sweepGossips(period);
    } catch (Exception ex) {
      LOGGER.warn("Exception at doSpreadGossip[{}]: {}", period, ex.getMessage(), ex);
    }
  }

  private void spreadGossipsTo(long period, Member member) {
    // Select gossips to send
    List<Gossip> gossips = selectGossipsToSend(period, member);
    if (gossips.isEmpty()) {
      return; // nothing to spread
    }

    // Send gossip request
    Address address = member.address();

    gossips
        .stream()
        .map(this::buildGossipRequestMessage)
        .forEach(
            message ->
                transport
                    .send(address, message)
                    .subscribe(
                        null,
                        ex ->
                            LOGGER.debug(
                                "Failed to send GossipReq[{}]: {} to {}, cause: {}",
                                period,
                                message,
                                address,
                                ex.toString())));
  }

  private List<Gossip> selectGossipsToSend(long period, Member member) {
    int periodsToSpread =
        ClusterMath.gossipPeriodsToSpread(config.getGossipRepeatMult(), remoteMembers.size() + 1);
    return gossips
        .values()
        .stream()
        .filter(
            gossipState -> gossipState.infectionPeriod() + periodsToSpread >= period) // max rounds
        .filter(gossipState -> !gossipState.isInfected(member.id())) // already infected
        .map(GossipState::gossip)
        .collect(Collectors.toList());
  }

  private List<Member> selectGossipMembers() {
    int gossipFanout = config.getGossipFanout();
    if (remoteMembers.size() < gossipFanout) { // select all
      return remoteMembers;
    } else { // select random members
      // Shuffle members initially and once reached top bound
      if (remoteMembersIndex < 0 || remoteMembersIndex + gossipFanout > remoteMembers.size()) {
        Collections.shuffle(remoteMembers);
        remoteMembersIndex = 0;
      }

      // Select members
      List<Member> selectedMembers =
          gossipFanout == 1
              ? Collections.singletonList(remoteMembers.get(remoteMembersIndex))
              : remoteMembers.subList(remoteMembersIndex, remoteMembersIndex + gossipFanout);

      // Increment index and return result
      remoteMembersIndex += gossipFanout;
      return selectedMembers;
    }
  }

  private Message buildGossipRequestMessage(Gossip gossip) {
    GossipRequest gossipRequest = new GossipRequest(gossip, localMember.id());
    return Message.withData(gossipRequest)
        .qualifier(GOSSIP_REQ)
        .sender(localMember.address())
        .build();
  }

  private void sweepGossips(long period) {
    // Select gossips to sweep
    int periodsToSweep =
        ClusterMath.gossipPeriodsToSweep(config.getGossipRepeatMult(), remoteMembers.size() + 1);
    Set<GossipState> gossipsToRemove =
        gossips
            .values()
            .stream()
            .filter(gossipState -> period > gossipState.infectionPeriod() + periodsToSweep)
            .collect(Collectors.toSet());

    // Check if anything selected
    if (gossipsToRemove.isEmpty()) {
      return; // nothing to sweep
    }

    // Sweep gossips
    LOGGER.debug("Sweep gossips[{}]: {}", period, gossipsToRemove);
    for (GossipState gossipState : gossipsToRemove) {
      gossips.remove(gossipState.gossip().gossipId());
      MonoSink<String> sink = futures.remove(gossipState.gossip().gossipId());
      if (sink != null) {
        sink.success(gossipState.gossip().gossipId());
      }
    }
  }

  //......

}
  • The doSpreadGossip method first increments the currentPeriod, then executes selectGossipMembers, traverses the member to execute spreadGossipsTo, and finally executes sweepGossips
  • The SelectgossipFanout method randomly selects gossipFanout member according to Gossip Fanout configuration. remoteMembersIndex is maintained here, specifically subList the remoteMembers. When remoteMembersIndex is less than 0 or remoteMembersIndex+gossipFanout > remotemembers.size (), collections.shuffle (remotemembers) and reset remoteMembersIndex to 0, followed by adding gossip fanout to remotemembersindex
  • The spreadgossipsTo method first executes selectGossipsToSend to obtain Gossips to send, then constructs GOSSIP_REQ message through buildGossipRequestMessage, and finally sends it through transport.send method
  • The sweepgossips method selects periodsToSweep, and then removes from Gossips the gossipState of Period > Gossip State.

Summary

  • GossipProtocol interface defines start, stop, spread, listen methods; GossipProtocol interface is implemented by Gossip Protocol, which maintains the gossipId named gossips, the map of GossipState, and the list of remoteMembers.
  • GossipProtocolImpl’s constructor subscribes to the member; hipProcessor, triggering the onMemberEvent method, which adds or removes members to remoteMembers according to the MembershipEvent; Subscribed to transport.listen (), filtered out gossipReq and triggered ongossipReq method. The method merges gossips of GossipRequest into local gossips, and sends message for new Gossip to sink, maintains gossipState of Gossip, and adds the requested memberId to infected; The spread method places the message into the local gossips
  • The start method of GossipProtocolImpl executes the doSpreadGossip method every gossipInterval; Spread method creates Gossip through createAndPutGossip and puts it into gossips. The doSpreadGossip method first increments the currentPeriod, then executes selectGossipMembers, traverses the member to execute spreadGossipsTo, and finally executes sweepGossips

GossipProtocolImpl has registered onMemberEvent and onGossipReq here, where onMemberEvent is used to monitor the MembershipEvent and maintain the remoteMembers list according to the event; OnGossipReq is a GossipReq message sent by the doSpreadGossip method of other member, and merges the gossips of the message into the local Gossips; However, the doSpreadGossip method is executed at every gossipInterval. gossipFanout member are randomly selected according to gossipFanout configuration, and then gossips to be sent are selected for spread (OnGossipReq and spread methods change gossips, while doSpreadGossip triggered by every gossipInterval selects the message to be spread from gossips to send.)

doc