|
| 1 | +package org.jgroups.tests; |
| 2 | + |
| 3 | +import org.jgroups.*; |
| 4 | +import org.jgroups.protocols.FRAG4; |
| 5 | +import org.jgroups.protocols.TP; |
| 6 | +import org.jgroups.protocols.UFC; |
| 7 | +import org.jgroups.protocols.pbcast.GMS; |
| 8 | +import org.jgroups.stack.Protocol; |
| 9 | +import org.jgroups.stack.ProtocolStack; |
| 10 | +import org.jgroups.util.Util; |
| 11 | +import org.testng.annotations.Test; |
| 12 | + |
| 13 | +import java.util.ArrayList; |
| 14 | +import java.util.List; |
| 15 | +import java.util.stream.Collectors; |
| 16 | +import java.util.stream.Stream; |
| 17 | + |
| 18 | +/** |
| 19 | + * Single-JVM reproducer for JGroups issue 1048: a stale view reaches FlowControl after a newer one, |
| 20 | + * and evicts a member that is still in the cluster. |
| 21 | + * |
| 22 | + * WHAT IT SHOWS |
| 23 | + * |
| 24 | + * GMS.installView() guards against stale views correctly, but it sets `view` under the lock and |
| 25 | + * dispatches after releasing it (5.5.1 lines 691-699): |
| 26 | + * |
| 27 | + * finally { lock.unlock(); } |
| 28 | + * down_prot.down(view_event); |
| 29 | + * up_prot.up(view_event); |
| 30 | + * |
| 31 | + * So installation order and dispatch order are separate. Two threads can be inside installView() |
| 32 | + * for different views, both passing the guard legitimately: the joining application thread inside |
| 33 | + * connect() (ClientGmsImpl.installViewIfValidJoinRsp -> GMS.installView) and a JGroups thread |
| 34 | + * handling the coordinator's next view. If anything below GMS is slow for the older view's |
| 35 | + * down_prot.down(), the newer view completes its whole dispatch first and the older one arrives at |
| 36 | + * FlowControl afterwards. FlowControl.up() has no view-id check, so handleViewChange() runs with |
| 37 | + * the older member list and retainAll() drops a live member. handleCreditRequest() then does |
| 38 | + * map.get(sender), finds null, and returns without replenishing: that peer's credit window drains |
| 39 | + * one way and never refills. |
| 40 | + * |
| 41 | + * The only thing simulated here is "something below GMS is slow", by SlowViewDown below. In |
| 42 | + * production that was FD_SOCK2/TCP setting up connections to a member that was itself still |
| 43 | + * starting; measured stalls there were 6.0s (four occurrences) and 8.03s. |
| 44 | + * |
| 45 | + * JIRA https://redhat.atlassian.net/browse/JGRP-3036 |
| 46 | + */ |
| 47 | +@Test(groups=Global.FUNCTIONAL) |
| 48 | +public class Repro1048 { |
| 49 | + static final String CLUSTER="repro1048"; |
| 50 | + |
| 51 | + public void test1048() throws Exception { |
| 52 | + JChannel a=channel("A", false); |
| 53 | + JChannel b=channel("B", true); |
| 54 | + JChannel c=channel("C", false); |
| 55 | + try { |
| 56 | + a.connect(CLUSTER); |
| 57 | + say("A connected, view=%s", a.getView()); |
| 58 | + |
| 59 | + // B joins on an APPLICATION thread, which is what connect() is in a real deployment. |
| 60 | + // Its first view's down-dispatch is then held, so connect() does not return yet. |
| 61 | + Thread joiner=new Thread(() -> { |
| 62 | + try { |
| 63 | + b.connect(CLUSTER); |
| 64 | + } |
| 65 | + catch(Exception e) { |
| 66 | + e.printStackTrace(); |
| 67 | + } |
| 68 | + }, "app-joiner-B"); |
| 69 | + joiner.start(); |
| 70 | + |
| 71 | + // Let B install its first view and enter the stall, then let C join. The coordinator moves to a 3-member |
| 72 | + // view and B applies it on a JGroups thread, while B's own 2-member view is still undelivered. |
| 73 | + Util.sleep(2000); |
| 74 | + c.connect(CLUSTER); |
| 75 | + say("C connected, view=%s", c.getView()); |
| 76 | + |
| 77 | + joiner.join(); |
| 78 | + say("B connect() returned, view=%s", b.getView()); |
| 79 | + Util.sleep(1000); |
| 80 | + |
| 81 | + System.out.printf("\n-------------- channel views:\n%s\n", |
| 82 | + Stream.of(a,b,c).map(ch -> String.format("%s: %s", ch.address(), ch.view())) |
| 83 | + .collect(Collectors.joining("\n"))); |
| 84 | + |
| 85 | + System.out.printf("\n-------------- GMS views:\n%s\n\n", |
| 86 | + Stream.of(a,b,c) |
| 87 | + .map(ch -> ch.stack().findProtocol(GMS.class)) |
| 88 | + .map(gms -> String.format("%s: %s", ((GMS)gms).addr(), ((GMS)gms).view())) |
| 89 | + .collect(Collectors.joining("\n"))); |
| 90 | + |
| 91 | + // The coordinator's view is authoritative: every member in it is alive. |
| 92 | + View cluster = a.getView(); |
| 93 | + List<Address> members = cluster.getMembers(); |
| 94 | + UFC ufc = b.getProtocolStack().findProtocol(UFC.class); |
| 95 | + String senders = ufc.printSenderCredits(); |
| 96 | + String recv = ufc.printReceiverCredits(); |
| 97 | + |
| 98 | + List<Address> missing=new ArrayList<>(); |
| 99 | + for(Address m: members) |
| 100 | + if(!inMap(senders, m) || !inMap(recv, m)) |
| 101 | + missing.add(m); |
| 102 | + |
| 103 | + say("cluster view (from coordinator A): %s", cluster); |
| 104 | + say("B's channel view: %s", b.getView()); |
| 105 | + say("B's UFC credits:%n%s", ufc.printCredits()); |
| 106 | + |
| 107 | + if(missing.isEmpty()) { |
| 108 | + say("NOT REPRODUCED: B's credit maps contain every member of %s", cluster.getViewId()); |
| 109 | + return; |
| 110 | + } |
| 111 | + say("REPRODUCED: %s is in view %s but missing from B's UFC maps." |
| 112 | + + " handleCreditRequest() will drop its credit requests, and its window drains one way.", |
| 113 | + missing, cluster.getViewId()); |
| 114 | + assert false : String.format("REPRODUCED: %s is in view %s but missing from B's UFC maps." |
| 115 | + + " handleCreditRequest() will drop its credit requests, and its window drains one way.", |
| 116 | + missing, cluster.getViewId()); |
| 117 | + } |
| 118 | + finally { |
| 119 | + Util.close(c, b, a); |
| 120 | + } |
| 121 | + } |
| 122 | + |
| 123 | + /** Minimal TCP stack. SlowViewDown sits directly above the transport, where a slow protocol is. */ |
| 124 | + static JChannel channel(String name, boolean slow) throws Exception { |
| 125 | + JChannel ch=new JChannel(Util.getTestStack(new UFC().setMaxCredits(2_000_000), new FRAG4())).name(name); |
| 126 | + if(slow) { |
| 127 | + SlowViewDown sv=new SlowViewDown().delay(4000); |
| 128 | + ch.stack().insertProtocol(sv, ProtocolStack.Position.ABOVE, TP.class); |
| 129 | + } |
| 130 | + return ch; |
| 131 | + } |
| 132 | + |
| 133 | + /** printMap() emits "<addr>: <credits>" per line. */ |
| 134 | + static boolean inMap(String credit_map, Address mbr) { |
| 135 | + for(String line: credit_map.split("\n")) { |
| 136 | + int idx=line.lastIndexOf(':'); |
| 137 | + if(idx > 0 && line.substring(0, idx).trim().equals(mbr.toString())) |
| 138 | + return true; |
| 139 | + } |
| 140 | + return false; |
| 141 | + } |
| 142 | + |
| 143 | + static void say(String fmt, Object... args) { |
| 144 | + //noinspection StringConcatenationInFormatCall |
| 145 | + System.out.printf("%tT.%<tL " + fmt + "%n", prepend(System.currentTimeMillis(), args)); |
| 146 | + } |
| 147 | + |
| 148 | + static Object[] prepend(Object first, Object[] rest) { |
| 149 | + Object[] all=new Object[rest.length + 1]; |
| 150 | + all[0]=first; |
| 151 | + System.arraycopy(rest, 0, all, 1, rest.length); |
| 152 | + return all; |
| 153 | + } |
| 154 | + |
| 155 | + /** |
| 156 | + * Holds the down-dispatch of a view, and only when the dispatching thread is not a JGroups |
| 157 | + * stack thread. That restriction matters: delaying a stack thread blocks that sender's message |
| 158 | + * queue, so the newer view would never arrive and nothing would reproduce. |
| 159 | + */ |
| 160 | + public static class SlowViewDown extends Protocol { |
| 161 | + protected long delay=8000; |
| 162 | + |
| 163 | + public SlowViewDown delay(long d) {delay=d; return this;} |
| 164 | + |
| 165 | + public Object down(Event evt) { |
| 166 | + if(evt.getType() == Event.VIEW_CHANGE) { |
| 167 | + View v = evt.getArg(); |
| 168 | + String thread = Thread.currentThread().getName(); |
| 169 | + boolean off_stack=!thread.startsWith("jgroups-") && !thread.startsWith("thread-"); |
| 170 | + if(v.getViewId().getId() <= 1 && off_stack) { |
| 171 | + say(" [SlowViewDown] holding down-dispatch of %s for %dms on thread %s", |
| 172 | + v.getViewId(), delay, thread); |
| 173 | + Util.sleep(delay); |
| 174 | + say(" [SlowViewDown] releasing %s", v.getViewId()); |
| 175 | + } |
| 176 | + } |
| 177 | + return down_prot.down(evt); |
| 178 | + } |
| 179 | + } |
| 180 | +} |
0 commit comments