Skip to content

Commit

Permalink
FABJ-383 Garbage collection orderer peer connection.
Browse files Browse the repository at this point in the history 
PS
06 Minor logging updates.
07 Logging changes.

Change-Id: I1e6ed709b43b987ad573da15af63a36b492c8850
Signed-off-by: rickr <cr22rc@gmail.com>
  • Loading branch information
@cr22rc
cr22rc committed Nov 8, 2018
1 parent 8044bac commit 1eb8e0a
Show file tree
Hide file tree
Showing 25 changed files with 614 additions and 327 deletions.
129 changes: 90 additions & 39 deletions src/main/java/org/hyperledger/fabric/sdk/Channel.java
View file

Large diffs are not rendered by default.

66 changes: 56 additions & 10 deletions src/main/java/org/hyperledger/fabric/sdk/EndorserClient.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,6 @@
import java.util.concurrent.TimeUnit;

import com.google.common.util.concurrent.ListenableFuture;
import io.grpc.ConnectivityState;
import io.grpc.ManagedChannel;
import io.grpc.ManagedChannelBuilder;
import org.apache.commons.logging.Log;
Expand All @@ -28,30 +27,50 @@
import org.hyperledger.fabric.protos.peer.FabricProposal;
import org.hyperledger.fabric.protos.peer.FabricProposalResponse;
import org.hyperledger.fabric.sdk.exception.PeerException;
import org.hyperledger.fabric.sdk.helper.Config;

import static java.lang.String.format;

/**
* Sample client code that makes gRPC calls to the server.
*/
class EndorserClient {
private static final Config config = Config.getConfig();
private static final Log logger = LogFactory.getLog(EndorserClient.class);
private static final boolean IS_TRACE_LEVEL = logger.isTraceEnabled();

// private final String channelName;
// private final String name;
// private final String url;
private ManagedChannel managedChannel;
private EndorserGrpc.EndorserFutureStub futureStub;
DiscoveryGrpc.DiscoveryFutureStub discoveryFutureStub;
private DiscoveryGrpc.DiscoveryFutureStub discoveryFutureStub;
private boolean shutdown = false;
private final String toString;

/**
* Construct client for accessing Peer server using the existing channel.
*
* @param channelBuilder The ChannelBuilder to build the endorser client
*/
EndorserClient(ManagedChannelBuilder<?> channelBuilder) {
EndorserClient(String channelName, String name, String url, ManagedChannelBuilder<?> channelBuilder) {
managedChannel = channelBuilder.build();
futureStub = EndorserGrpc.newFutureStub(managedChannel);
discoveryFutureStub = DiscoveryGrpc.newFutureStub(managedChannel);
toString = "EndorserClient{" + "id: " + config.getNextID() + ", channel: " + channelName + ", name:" + name + ", url: " + url + "}";
logger.trace("Created " + toString());

}

@Override
public String toString() {
return toString;
}

synchronized void shutdown(boolean force) {
if (IS_TRACE_LEVEL) {
logger.trace(format("%s shutdown called force: %b, shutdown: %b, managedChannel: %s", toString(), force, shutdown, "" + managedChannel));
}
if (shutdown) {
return;
}
Expand All @@ -60,49 +79,76 @@ synchronized void shutdown(boolean force) {
// let all referenced resource finalize
managedChannel = null;
discoveryFutureStub = null;

futureStub = null;

if (lchannel == null) {
return;
}
if (force) {
lchannel.shutdownNow();

try {
lchannel.shutdownNow();
} catch (Exception e) {
logger.warn(e);
}

} else {
boolean isTerminated = false;

try {
isTerminated = lchannel.shutdown().awaitTermination(3, TimeUnit.SECONDS);
} catch (Exception e) {
logger.debug(e); //best effort
logger.debug(toString(), e); //best effort
}
if (!isTerminated) {
lchannel.shutdownNow();
try {
lchannel.shutdownNow();
} catch (Exception e) {
logger.warn(toString(), e);
}
}
}
}

public ListenableFuture<FabricProposalResponse.ProposalResponse> sendProposalAsync(FabricProposal.SignedProposal proposal) throws PeerException {
if (shutdown) {
throw new PeerException("Shutdown");
throw new PeerException("Shutdown " + toString());
}
return futureStub.processProposal(proposal);
}

public ListenableFuture<Protocol.Response> sendDiscoveryRequestAsync(Protocol.SignedRequest signedRequest) throws PeerException {
if (shutdown) {
throw new PeerException("Shutdown");
throw new PeerException("Shutdown " + toString());
}
return discoveryFutureStub.discover(signedRequest);
}

boolean isChannelActive() {
ManagedChannel lchannel = managedChannel;
return lchannel != null && !lchannel.isShutdown() && !lchannel.isTerminated() && ConnectivityState.READY.equals(lchannel.getState(true));
if (null == lchannel) {

logger.trace(toString() + "Grpc channel needs creation.");

return false;
}

final boolean isTerminated = lchannel.isTerminated();
final boolean isShutdown = lchannel.isShutdown();
final boolean ret = !lchannel.isShutdown() && !isTerminated; // && ConnectivityState.READY.equals(lchannel.getState(true));
if (IS_TRACE_LEVEL) {
logger.trace(format("%s grpc channel isActive: %b, isShutdown: %b, isTerminated: %b, state: %s ", toString(), ret, isShutdown, isTerminated, "" + lchannel.getState(false)));
}

return ret;
}

@Override
public void finalize() {
if (!shutdown) {
logger.warn(toString() + " finalized not shutdown is Active" + isChannelActive());
}

shutdown(true);
}
}
82 changes: 64 additions & 18 deletions src/main/java/org/hyperledger/fabric/sdk/EventHub.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,8 @@

package org.hyperledger.fabric.sdk;

import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.Serializable;
import java.util.ArrayList;
import java.util.Properties;
Expand Down Expand Up @@ -52,8 +54,11 @@

public class EventHub implements Serializable {
private static final long serialVersionUID = 2882609588201108148L;
private static final Log logger = LogFactory.getLog(EventHub.class);
private static final Config config = Config.getConfig();
private transient String id = config.getNextID();
private static final Log logger = LogFactory.getLog(EventHub.class);
private static final boolean IS_TRACE_LEVEL = logger.isTraceEnabled();

private static final long EVENTHUB_CONNECTION_WAIT_TIME = config.getEventHubConnectionWaitTime();
private static final long EVENTHUB_RECONNECTION_WARNING_RATE = config.getEventHubReconnectionWarningRate();

Expand All @@ -78,6 +83,7 @@ public class EventHub implements Serializable {
private transient long reconnectCount;
private transient long lastBlockNumber;
private transient BlockEvent lastBlockEvent;
private String channelName;

/**
* Get disconnected time.
Expand All @@ -99,6 +105,22 @@ public boolean isConnected() {
return connected;
}

String getStatus() {

StringBuilder sb = new StringBuilder(1000);
sb.append(toString()).append(", connected: ").append(connected);
ManagedChannel lmanagedChannel = managedChannel;
if (lmanagedChannel == null) {
sb.append("managedChannel: null");
} else {
sb.append(", isShutdown: ").append(lmanagedChannel.isShutdown());
sb.append(", isTerminated: ").append(lmanagedChannel.isTerminated());
sb.append(", state: ").append("" + lmanagedChannel.getState(false));
}

return sb.toString();
}

/**
* Get last connect time.
*
Expand Down Expand Up @@ -136,6 +158,7 @@ public long getLastConnectedAttempt() {
this.name = name;
this.executorService = executorService;
this.properties = properties == null ? null : (Properties) properties.clone(); //keep our own copy.
logger.debug("Created " + toString());
}

/**
Expand Down Expand Up @@ -186,7 +209,7 @@ synchronized boolean connect(final TransactionContext transactionContext, final

final CountDownLatch finishLatch = new CountDownLatch(1);

logger.debug(format("EventHub %s is connecting.", name));
logger.debug(format("%s is connecting.", toString()));

lastConnectedAttempt = System.currentTimeMillis();

Expand All @@ -203,12 +226,14 @@ synchronized boolean connect(final TransactionContext transactionContext, final
@Override
public void onNext(PeerEvents.Event event) {

logger.debug(format("EventHub %s got event type: %s", EventHub.this.name, event.getEventCase().name()));
logger.debug(format("%s got event type: %s", EventHub.this.toString(), event.getEventCase().name()));

if (event.getEventCase() == PeerEvents.Event.EventCase.BLOCK) {
try {

BlockEvent blockEvent = new BlockEvent(EventHub.this, event);

logger.trace(format("%s got block number: %d", EventHub.this.toString(), blockEvent.getBlockNumber()));
setLastBlockSeen(blockEvent);

eventQue.addBEvent(blockEvent); //add to channel queue
Expand All @@ -220,14 +245,17 @@ public void onNext(PeerEvents.Event event) {
} else if (event.getEventCase() == PeerEvents.Event.EventCase.REGISTER) {

if (reconnectCount > 1) {
logger.info(format("Eventhub %s has reconnecting after %d attempts", name, reconnectCount));
logger.info(format("%s has reconnecting after %d attempts", EventHub.this.toString(), reconnectCount));
}

connected = true;
connectedTime = System.currentTimeMillis();
reconnectCount = 0L;

finishLatch.countDown();
} else {
logger.error(format("%s got a unexpected block type: %s",
EventHub.this.toString(), event.getEventCase().name()));
}
}

Expand Down Expand Up @@ -271,7 +299,7 @@ public void onError(Throwable t) {
try {
reconnect();
} catch (Exception e) {
logger.warn(format("Eventhub %s Failed shutdown msg: %s", EventHub.this.name, e.getMessage()));
logger.warn(format("%s Failed shutdown msg: %s", EventHub.this.toString(), e.getMessage()));
}

}
Expand Down Expand Up @@ -300,17 +328,17 @@ public void onCompleted() {

if (!reconnection && !finishLatch.await(EVENTHUB_CONNECTION_WAIT_TIME, TimeUnit.MILLISECONDS)) {

logger.warn(format("EventHub %s failed to connect in %s ms.", name, EVENTHUB_CONNECTION_WAIT_TIME));
logger.warn(format("%s failed to connect in %s ms.", toString(), EVENTHUB_CONNECTION_WAIT_TIME));

} else {
logger.trace(format("Eventhub %s Done waiting for reply!", name));
logger.trace(format("%s done waiting for reply!", toString()));
}

} catch (InterruptedException e) {
logger.error(e);
}

logger.debug(format("Eventhub %s connect is done with connect status: %b ", name, connected));
logger.debug(format("%s connect is done with connect status: %b ", toString(), connected));

if (connected) {
eventStream = eventStreamLocal;
Expand Down Expand Up @@ -383,16 +411,19 @@ void setEventQue(Channel.ChannelEventQue eventQue) {

@Override
public String toString() {
return "EventHub:" + getName();
return "EventHub{" + "id: " + id + ", name: " + getName() + ", channelName: " + channelName + ", url: " + getUrl() + "}";
}

public void shutdown() {
public synchronized void shutdown() {
if (shutdown) {
return;
}
logger.trace(toString() + " being shutdown.");
shutdown = true;
lastBlockEvent = null;
lastBlockNumber = 0;
connected = false;
disconnectedHandler = null;
channel = null;
eventStream = null;
final ManagedChannel lmanagedChannel = managedChannel;
managedChannel = null;
Expand All @@ -406,20 +437,23 @@ void setChannel(Channel channel) throws InvalidArgumentException {
throw new InvalidArgumentException("setChannel Channel can not be null");
}

if (null != this.channel) {
if (null != channelName) {
throw new InvalidArgumentException(format("Can not add event hub %s to channel %s because it already belongs to channel %s.",
name, channel.getName(), this.channel.getName()));
name, channel.getName(), channelName));
}

this.channel = channel;
logger.debug(toString() + " set to channel: " + channel);
this.channelName = channel.getName();
}

synchronized void setLastBlockSeen(BlockEvent lastBlockSeen) {
private synchronized void setLastBlockSeen(BlockEvent lastBlockSeen) {
long newLastBlockNumber = lastBlockSeen.getBlockNumber();
// overkill but make sure.
if (lastBlockNumber < newLastBlockNumber) {
lastBlockNumber = newLastBlockNumber;
this.lastBlockEvent = lastBlockSeen;
if (IS_TRACE_LEVEL) {
logger.trace(toString() + " last block seen: " + lastBlockNumber);
}
}
}

Expand All @@ -446,7 +480,7 @@ public interface EventHubDisconnected {
@Override
public synchronized void disconnected(final EventHub eventHub) {
if (reconnectCount == 1) {
logger.warn(format("Channel %s detected disconnect on event hub %s (%s)", channel.getName(), eventHub.toString(), url));
logger.warn(format("%s detected disconnect.", eventHub.toString()));
}

executorService.execute(() -> {
Expand All @@ -455,7 +489,7 @@ public synchronized void disconnected(final EventHub eventHub) {
Thread.sleep(500);

if (transactionContext == null) {
logger.warn("Eventhub reconnect failed with no user context");
logger.warn(EventHub.this.toString() + " reconnect failed with no user context");
return;
}

Expand Down Expand Up @@ -485,4 +519,16 @@ public EventHubDisconnected setEventHubDisconnectedHandler(EventHubDisconnected
return ret;
}

private void readObject(ObjectInputStream in) throws IOException, ClassNotFoundException {
in.defaultReadObject();
id = config.getNextID();
}

public void finalize() throws Throwable {
logger.trace(format("%s finalized", toString()));
shutdown();
super.finalize();

}

}
3 changes: 1 addition & 2 deletions src/main/java/org/hyperledger/fabric/sdk/HFClient.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -51,7 +51,7 @@
import static org.hyperledger.fabric.sdk.User.userContextCheck;

public class HFClient {
private static final Config config = Config.getConfig();
private static final Config config = Config.getConfig(); // never remove this! config needs to load first.

private CryptoSuite cryptoSuite;
protected final ExecutorService executorService;
Expand All @@ -66,7 +66,6 @@ public class HFClient {
}
}


ExecutorService getExecutorService() {
return executorService;
}
Expand Down
Loading

0 comments on commit 1eb8e0a

Please sign in to comment.