Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Added Discovery.max_rank_to_reply: if TP.use_ip_addrs is true, then…
… only members with a rank smaller than max_rank_to_reply respond to initial membership discovery requests; others drop it (https://issues.jboss.org/browse/JGRP-2098) - Added field only_coords_send_discovery_rsps to Discovery - Added View.newMembers(View old, View new_view) - Added initial_discovery to PingHeader to let responders decide whether or not to reply - Added View.sameViews() plus unit test - Merger now rejects a merge if all views passed to it are equal (same ViewId)
- Loading branch information
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Original file line | Diff line number | Diff line change |
|---|---|---|---|
|
|
@@ -40,20 +40,17 @@ public abstract class Discovery extends Protocol { | ||
|
|
|
||
| @Property(description="Whether or not to return the entire logical-physical address cache mappings on a " + | @Property(description="Whether or not to return the entire logical-physical address cache mappings on a " + | ||
| "discovery request, or not.") | "discovery request, or not.") | ||
| protected boolean return_entire_cache; | |||
|
|
|
||
| @Property(description="If greater than 0, we'll wait a random number of milliseconds in range [0..stagger_timeout] " + | @Property(description="If greater than 0, we'll wait a random number of milliseconds in range [0..stagger_timeout] " + | ||
| "before sending a discovery response. This prevents traffic spikes in large clusters when everyone sends their " + | "before sending a discovery response. This prevents traffic spikes in large clusters when everyone sends their " + | ||
| "discovery response at the same time") | "discovery response at the same time") | ||
| protected long stagger_timeout; | protected long stagger_timeout; | ||
|
|
|
||
|
|
|
||
| @Property(description="If a persistent disk cache (PDC) is present, combine the discovery results with the " + | @Property(description="If a persistent disk cache (PDC) is present, combine the discovery results with the " + | ||
| "contents of the disk cache before returning the results") | "contents of the disk cache before returning the results") | ||
| protected boolean use_disk_cache; | |||
|
|
|
||
| @Property(description="Max size of the member list shipped with a discovery request. If we have more, the " + | @Property(description="Max size of the member list shipped with a discovery request. If we have more, the " + | ||
| "mbrs field in the discovery request header is nulled and members return the entire membership, " + | "mbrs field in the discovery request header is nulled and members return the entire membership, " + | ||
|
|
@@ -65,57 +62,57 @@ public abstract class Discovery extends Protocol { | ||
|
|
|
||
| @Property(description="If true then the discovery is done on a separate timer thread. Should be set to true when " + | @Property(description="If true then the discovery is done on a separate timer thread. Should be set to true when " + | ||
| "discovery is blocking and/or takes more than a few milliseconds") | "discovery is blocking and/or takes more than a few milliseconds") | ||
| protected boolean async_discovery; | |||
|
|
|
||
| @Property(description="If enabled, use a separate thread for every discovery request. Can be used with or without " + | @Property(description="If enabled, use a separate thread for every discovery request. Can be used with or without " + | ||
| "async_discovery") | "async_discovery") | ||
| protected boolean async_discovery_use_separate_thread_per_request; | protected boolean async_discovery_use_separate_thread_per_request; | ||
|
|
|
||
| @Property(description="When a new node joins, and we have a static discovery protocol (TCPPING), then send the " + | @Property(description="When a new node joins, and we have a static discovery protocol (TCPPING), then send the " + | ||
| "contents of the discovery cache to new and existing members if true (and we're the coord). Addresses JGRP-1903") | "contents of the discovery cache to new and existing members if true (and we're the coord). Addresses JGRP-1903") | ||
| protected boolean send_cache_on_join; | |||
|
|
|
||
| @Property(description="The max rank of this member to respond to discovery requests, e.g. if " + | |||
| "max_rank_to_reply=2 in {A,B,C,D,E}, only A (rank 1) and B (rank 2) will reply. A value <= 0 means " + | |||
| "everybody will reply. This attribute is ignored if TP.use_ip_addrs is false.") | |||
| protected int max_rank_to_reply; | |||
|
|
|
||
| /* --------------------------------------------- JMX ------------------------------------------------------ */ | /* --------------------------------------------- JMX ------------------------------------------------------ */ | ||
|
|
|
||
| @ManagedAttribute(description="Total number of discovery requests sent ") | @ManagedAttribute(description="Total number of discovery requests sent ") | ||
| protected int num_discovery_requests; | protected int num_discovery_requests; | ||
|
|
|
||
| /* --------------------------------------------- Fields ------------------------------------------------------ */ | /* --------------------------------------------- Fields ------------------------------------------------------ */ | ||
|
|
|
||
| protected volatile boolean is_server; | |||
| protected volatile boolean is_leaving; | |||
| protected TimeScheduler timer; | protected TimeScheduler timer; | ||
| protected volatile View view; | |||
| @ManagedAttribute(description="Whether this member is the current coordinator") | @ManagedAttribute(description="Whether this member is the current coordinator") | ||
| protected volatile boolean is_coord; | |||
| protected volatile Address local_addr; | |||
| protected volatile Address current_coord; | |||
| protected String cluster_name; | protected String cluster_name; | ||
| protected final Map<Long,Responses> ping_responses=new HashMap<>(); | protected final Map<Long,Responses> ping_responses=new HashMap<>(); | ||
| @ManagedAttribute(description="Whether the transport supports multicasting") | @ManagedAttribute(description="Whether the transport supports multicasting") | ||
| protected boolean transport_supports_multicasting=true; | protected boolean transport_supports_multicasting=true; | ||
| protected boolean use_ip_addrs; // caches TP.use_ip_addrs | |||
| @ManagedAttribute(description="True if sending a message can block at the transport level") | @ManagedAttribute(description="True if sending a message can block at the transport level") | ||
| protected boolean sends_can_block=true; | protected boolean sends_can_block=true; | ||
| protected static final byte[] WHITESPACE=" \t".getBytes(); | protected static final byte[] WHITESPACE=" \t".getBytes(); | ||
|
|
|
||
|
|
|
||
|
|
|
||
| public void init() throws Exception { | public void init() throws Exception { | ||
| TP tp=getTransport(); | |||
| timer=tp.getTimer(); | |||
| if(timer == null) | if(timer == null) | ||
| throw new Exception("timer cannot be retrieved from protocol stack"); | throw new Exception("timer cannot be retrieved from protocol stack"); | ||
| if(stagger_timeout < 0) | if(stagger_timeout < 0) | ||
| throw new IllegalArgumentException("stagger_timeout cannot be negative"); | throw new IllegalArgumentException("stagger_timeout cannot be negative"); | ||
| transport_supports_multicasting=tp.supportsMulticasting(); | |||
| sends_can_block=getTransport() instanceof TCP; // UDP and TCP_NIO2 won't block | sends_can_block=getTransport() instanceof TCP; // UDP and TCP_NIO2 won't block | ||
| use_ip_addrs=tp.getUseIpAddresses(); | |||
| } | } | ||
|
|
|
||
| public abstract boolean isDynamic(); | public abstract boolean isDynamic(); | ||
|
|
@@ -138,8 +135,6 @@ public void discoveryRequestReceived(Address sender, String logical_name, Physic | ||
| public Discovery returnEntireCache(boolean flag) {return_entire_cache=flag; return this;} | public Discovery returnEntireCache(boolean flag) {return_entire_cache=flag; return this;} | ||
| public long staggerTimeout() {return stagger_timeout;} | public long staggerTimeout() {return stagger_timeout;} | ||
| public Discovery staggerTimeout(long timeout) {stagger_timeout=timeout; return this;} | public Discovery staggerTimeout(long timeout) {stagger_timeout=timeout; return this;} | ||
| public boolean useDiskCache() {return use_disk_cache;} | public boolean useDiskCache() {return use_disk_cache;} | ||
| public Discovery useDiskCache(boolean flag) {use_disk_cache=flag; return this;} | public Discovery useDiskCache(boolean flag) {use_disk_cache=flag; return this;} | ||
| public Discovery discoveryRspExpiryTime(long t) {this.discovery_rsp_expiry_time=t; return this;} | public Discovery discoveryRspExpiryTime(long t) {this.discovery_rsp_expiry_time=t; return this;} | ||
|
|
@@ -163,10 +158,7 @@ protected boolean isMergeRunning() { | ||
|
|
|
||
| @ManagedOperation(description="Sends information about my cache to everyone but myself") | @ManagedOperation(description="Sends information about my cache to everyone but myself") | ||
| public void sendCacheInformation() { | public void sendCacheInformation() { | ||
| List<Address> current_members=new ArrayList<>(view.getMembers()); | |||
| disseminateDiscoveryInformation(current_members, null, current_members); | disseminateDiscoveryInformation(current_members, null, current_members); | ||
| } | } | ||
|
|
|
||
|
|
@@ -292,7 +284,7 @@ public Object up(Message msg) { | ||
| } | } | ||
|
|
|
||
| // add physical address and logical name of the discovery sender (if available) to the cache | // add physical address and logical name of the discovery sender (if available) to the cache | ||
| if(data != null) { // null if use_ip_addrs is true | |||
| addDiscoveryResponseToCaches(logical_addr, data.getLogicalName(), data.getPhysicalAddr()); | addDiscoveryResponseToCaches(logical_addr, data.getLogicalName(), data.getPhysicalAddr()); | ||
| discoveryRequestReceived(msg.getSrc(), data.getLogicalName(), data.getPhysicalAddr()); | discoveryRequestReceived(msg.getSrc(), data.getLogicalName(), data.getPhysicalAddr()); | ||
| addResponse(data, false); | addResponse(data, false); | ||
|
|
@@ -304,9 +296,9 @@ public Object up(Message msg) { | ||
| for(Map.Entry<Address,PhysicalAddress> entry: cache.entrySet()) { | for(Map.Entry<Address,PhysicalAddress> entry: cache.entrySet()) { | ||
| Address addr=entry.getKey(); | Address addr=entry.getKey(); | ||
| // JGRP-1492: only return our own address, and addresses in view. | // JGRP-1492: only return our own address, and addresses in view. | ||
| if(addr.equals(local_addr) || view.containsMember(addr)) { | |||
| PhysicalAddress physical_addr=entry.getValue(); | PhysicalAddress physical_addr=entry.getValue(); | ||
| sendDiscoveryResponse(addr, physical_addr, NameCache.get(addr), msg.getSrc(), is_coord); | |||
This comment has been minimized.
Sorry, something went wrong.
This comment has been minimized.
Sorry, something went wrong. 
belaban
Author
Owner
|
|||
| } | } | ||
| } | } | ||
| } | } | ||
|
|
@@ -315,11 +307,11 @@ public Object up(Message msg) { | ||
|
|
|
||
| // Only send a response if hdr.mbrs is not empty and contains myself. Otherwise always send my info | // Only send a response if hdr.mbrs is not empty and contains myself. Otherwise always send my info | ||
| Collection<? extends Address> mbrs=data != null? data.mbrs() : null; | Collection<? extends Address> mbrs=data != null? data.mbrs() : null; | ||
| boolean drop_because_of_rank=use_ip_addrs && max_rank_to_reply > 0 && hdr.initialDiscovery() && Util.getRank(view, local_addr) > max_rank_to_reply; | |||
| if(drop_because_of_rank || (mbrs != null && !mbrs.contains(local_addr))) | |||
| return null; | |||
| PhysicalAddress physical_addr=(PhysicalAddress)down(new Event(Event.GET_PHYSICAL_ADDRESS, local_addr)); | |||
| sendDiscoveryResponse(local_addr, physical_addr, NameCache.get(local_addr), msg.getSrc(), is_coord); | |||
| return null; | return null; | ||
|
|
|
||
| case PingHeader.GET_MBRS_RSP: | case PingHeader.GET_MBRS_RSP: | ||
|
|
@@ -353,29 +345,19 @@ public Object down(Event evt) { | ||
| return findMembers(null, true, false); // triggered by JOIN process (ClientGmsImpl) | return findMembers(null, true, false); // triggered by JOIN process (ClientGmsImpl) | ||
|
|
|
||
| case Event.FIND_MBRS: | case Event.FIND_MBRS: | ||
| return findMembers(evt.getArg(), false, false); // triggered by MERGE3 | |||
|
|
|
||
| case Event.VIEW_CHANGE: | case Event.VIEW_CHANGE: | ||
| View old_view=view; | View old_view=view; | ||
| view=evt.getArg(); | view=evt.getArg(); | ||
| current_coord=view.getCoord(); | |||
| is_coord=current_coord != null && local_addr != null && current_coord.equals(local_addr); | is_coord=current_coord != null && local_addr != null && current_coord.equals(local_addr); | ||
| Object retval=down_prot.down(evt); | Object retval=down_prot.down(evt); | ||
| if(send_cache_on_join && !isDynamic() && is_coord) { | if(send_cache_on_join && !isDynamic() && is_coord) { | ||
| List<Address> curr_mbrs, left_mbrs, new_mbrs; | List<Address> curr_mbrs, left_mbrs, new_mbrs; | ||
| curr_mbrs=new ArrayList<>(view.getMembers()); | |||
| left_mbrs=View.leftMembers(old_view, view); | |||
| new_mbrs=View.newMembers(old_view, view); | |||
| startCacheDissemination(curr_mbrs, left_mbrs, new_mbrs); // separate task | startCacheDissemination(curr_mbrs, left_mbrs, new_mbrs); // separate task | ||
| } | } | ||
| return retval; | return retval; | ||
|
|
|||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.
Why did you change isCoord(addr) to is_coord ?
If this node is a coordinator, this change will cause all PingData responses to be recognized as Coord and return_entire_cache will not work properly.