Skip to content

Commit

Permalink
GG-11655 - Fix merge
Browse files Browse the repository at this point in the history
  • Loading branch information
@dkarachentsev
dkarachentsev committed Nov 2, 2016
1 parent 92fff63 commit a62a013
Showing 1 changed file with 74 additions and 62 deletions.
136 changes: 74 additions & 62 deletions ...ore/src/main/java/org/apache/ignite/internal/processors/service/GridServiceProcessor.java
View file
Expand Up @@ -851,7 +851,7 @@ public <T> T serviceProxy(ClusterGroup prj, String name, Class<? super T> svcItf
} }
} }


return new GridServiceProxy<>(prj, name, svcItf, sticky, ctx).proxy(); return new GridServiceProxy<T>(prj, name, svcItf, sticky, ctx).proxy();
} }


/** /**
Expand Down Expand Up @@ -904,7 +904,7 @@ public <T> Collection<T> services(String name) {
* @param topVer Topology version. * @param topVer Topology version.
* @throws IgniteCheckedException If failed. * @throws IgniteCheckedException If failed.
*/ */
private void reassign(GridServiceDeployment dep, long topVer) throws IgniteCheckedException { private void reassign(GridServiceDeployment dep, AffinityTopologyVersion topVer) throws IgniteCheckedException {
ServiceConfiguration cfg = dep.configuration(); ServiceConfiguration cfg = dep.configuration();


Object nodeFilter = cfg.getNodeFilter(); Object nodeFilter = cfg.getNodeFilter();
Expand All @@ -918,7 +918,7 @@ private void reassign(GridServiceDeployment dep, long topVer) throws IgniteCheck
Object affKey = cfg.getAffinityKey(); Object affKey = cfg.getAffinityKey();


while (true) { while (true) {
GridServiceAssignments assigns = new GridServiceAssignments(cfg, dep.nodeId(), topVer); GridServiceAssignments assigns = new GridServiceAssignments(cfg, dep.nodeId(), topVer.topologyVersion());


Collection<ClusterNode> nodes; Collection<ClusterNode> nodes;


Expand Down Expand Up @@ -948,7 +948,7 @@ private void reassign(GridServiceDeployment dep, long topVer) throws IgniteCheck
Map<UUID, Integer> cnts = new HashMap<>(); Map<UUID, Integer> cnts = new HashMap<>();


if (affKey != null) { if (affKey != null) {
ClusterNode n = ctx.affinity().mapKeyToNode(cacheName, affKey, new AffinityTopologyVersion(topVer)); ClusterNode n = ctx.affinity().mapKeyToNode(cacheName, affKey, topVer);


if (n != null) { if (n != null) {
int cnt = maxPerNodeCnt == 0 ? totalCnt == 0 ? 1 : totalCnt : maxPerNodeCnt; int cnt = maxPerNodeCnt == 0 ? totalCnt == 0 ? 1 : totalCnt : maxPerNodeCnt;
Expand Down Expand Up @@ -1180,7 +1180,7 @@ private Service copyAndInject(ServiceConfiguration cfg) throws IgniteCheckedExce
if (cfg instanceof LazyServiceConfiguration) { if (cfg instanceof LazyServiceConfiguration) {
byte[] bytes = ((LazyServiceConfiguration)cfg).serviceBytes(); byte[] bytes = ((LazyServiceConfiguration)cfg).serviceBytes();


Service srvc = m.unmarshal(bytes, U.resolveClassLoader(null, ctx.config())); Service srvc = U.unmarshal(m, bytes, U.resolveClassLoader(null, ctx.config()));


ctx.resource().inject(srvc); ctx.resource().inject(srvc);


Expand All @@ -1190,10 +1190,9 @@ private Service copyAndInject(ServiceConfiguration cfg) throws IgniteCheckedExce
Service svc = cfg.getService(); Service svc = cfg.getService();


try { try {
byte[] bytes = m.marshal(svc); byte[] bytes = U.marshal(m, svc);


Service cp = m.unmarshal(bytes, Service cp = U.unmarshal(m, bytes, U.resolveClassLoader(svc.getClass().getClassLoader(), ctx.config()));
U.resolveClassLoader(svc.getClass().getClassLoader(), ctx.config()));


ctx.resource().inject(cp); ctx.resource().inject(cp);


Expand Down Expand Up @@ -1268,8 +1267,8 @@ private Iterator<Cache.Entry<Object, Object>> serviceEntries(IgniteBiPredicate<O
ClusterNode oldestSrvNode = ClusterNode oldestSrvNode =
CU.oldestAliveCacheServerNode(cache.context().shared(), AffinityTopologyVersion.NONE); CU.oldestAliveCacheServerNode(cache.context().shared(), AffinityTopologyVersion.NONE);


if (oldestSrvNode == null) if (oldestSrvNode == null)
return F.emptyIterator(); return new GridEmptyIterator<>();


GridCacheQueryManager qryMgr = cache.context().queries(); GridCacheQueryManager qryMgr = cache.context().queries();


Expand Down Expand Up @@ -1455,7 +1454,7 @@ private void processDeployment(CacheEntryEvent<GridServiceDeploymentKey, GridSer
svcName.set(dep.configuration().getName()); svcName.set(dep.configuration().getName());


// Ignore other utility cache events. // Ignore other utility cache events.
long topVer = ctx.discovery().topologyVersion(); AffinityTopologyVersion topVer = ctx.discovery().topologyVersionEx();


ClusterNode oldest = U.oldest(ctx.discovery().nodes(topVer), null); ClusterNode oldest = U.oldest(ctx.discovery().nodes(topVer), null);


Expand Down Expand Up @@ -1506,60 +1505,60 @@ private void processDeployment(CacheEntryEvent<GridServiceDeploymentKey, GridSer
} }
} }


/** /**
* Deployment callback. * Deployment callback.
* *
* @param dep Service deployment. * @param dep Service deployment.
* @param topVer Topology version. * @param topVer Topology version.
*/ */
private void onDeployment(final GridServiceDeployment dep, final long topVer) { private void onDeployment(final GridServiceDeployment dep, final AffinityTopologyVersion topVer) {
// Retry forever. // Retry forever.
try { try {
long newTopVer = ctx.discovery().topologyVersion(); AffinityTopologyVersion newTopVer = ctx.discovery().topologyVersionEx();

// If topology version changed, reassignment will happen from topology event.
if (newTopVer == topVer)
reassign(dep, topVer);
}
catch (IgniteCheckedException e) {
if (!(e instanceof ClusterTopologyCheckedException))
log.error("Failed to do service reassignment (will retry): " + dep.configuration().getName(), e);

long newTopVer = ctx.discovery().topologyVersion();

if (newTopVer != topVer) {
assert newTopVer > topVer;


// Reassignment will happen from topology event. // If topology version changed, reassignment will happen from topology event.
return; if (newTopVer.equals(topVer))
reassign(dep, topVer);
} }
catch (IgniteCheckedException e) {
if (!(e instanceof ClusterTopologyCheckedException))
log.error("Failed to do service reassignment (will retry): " + dep.configuration().getName(), e);


ctx.timeout().addTimeoutObject(new GridTimeoutObject() { AffinityTopologyVersion newTopVer = ctx.discovery().topologyVersionEx();
private IgniteUuid id = IgniteUuid.randomUuid();


private long start = System.currentTimeMillis(); if (!newTopVer.equals(topVer)) {
assert newTopVer.compareTo(topVer) > 0;


@Override public IgniteUuid timeoutId() { // Reassignment will happen from topology event.
return id; return;
} }


@Override public long endTime() { ctx.timeout().addTimeoutObject(new GridTimeoutObject() {
return start + RETRY_TIMEOUT; private IgniteUuid id = IgniteUuid.randomUuid();
}


@Override public void onTimeout() { private long start = System.currentTimeMillis();
if (!busyLock.enterBusy())
return;


try { @Override public IgniteUuid timeoutId() {
// Try again. return id;
onDeployment(dep, topVer);
} }
finally {
busyLock.leaveBusy(); @Override public long endTime() {
return start + RETRY_TIMEOUT;
} }
}
}); @Override public void onTimeout() {
if (!busyLock.enterBusy())
return;

try {
// Try again.
onDeployment(dep, topVer);
}
finally {
busyLock.leaveBusy();
}
}
});
} }
} }


Expand All @@ -1568,16 +1567,28 @@ private void onDeployment(final GridServiceDeployment dep, final long topVer) {
*/ */
private class TopologyListener implements GridLocalEventListener { private class TopologyListener implements GridLocalEventListener {
/** {@inheritDoc} */ /** {@inheritDoc} */
@Override public void onEvent(final Event evt) { @Override public void onEvent(Event evt) {
if (!busyLock.enterBusy()) if (!busyLock.enterBusy())
return; return;


try { try {
final AffinityTopologyVersion topVer;

if (evt instanceof DiscoveryCustomEvent) {
DiscoveryCustomMessage msg = ((DiscoveryCustomEvent)evt).customMessage();

topVer = ((DiscoveryCustomEvent)evt).affinityTopologyVersion();

if (msg instanceof CacheAffinityChangeMessage) {
if (!((CacheAffinityChangeMessage)msg).exchangeNeeded())
return;
}
}
else
topVer = new AffinityTopologyVersion(((DiscoveryEvent)evt).topologyVersion(), 0);

depExe.submit(new BusyRunnable() { depExe.submit(new BusyRunnable() {
@Override public void run0() { @Override public void run0() {
AffinityTopologyVersion topVer =
new AffinityTopologyVersion(((DiscoveryEvent)evt).topologyVersion());

ClusterNode oldest = CU.oldestAliveCacheServerNode(cache.context().shared(), topVer); ClusterNode oldest = CU.oldestAliveCacheServerNode(cache.context().shared(), topVer);


if (oldest != null && oldest.isLocal()) { if (oldest != null && oldest.isLocal()) {
Expand Down Expand Up @@ -1612,7 +1623,7 @@ private class TopologyListener implements GridLocalEventListener {
ctx.cache().internalCache(UTILITY_CACHE_NAME).context().affinity(). ctx.cache().internalCache(UTILITY_CACHE_NAME).context().affinity().
affinityReadyFuture(topVer).get(); affinityReadyFuture(topVer).get();


reassign(dep, topVer.topologyVersion()); reassign(dep, topVer);
} }
catch (IgniteCheckedException ex) { catch (IgniteCheckedException ex) {
if (!(e instanceof ClusterTopologyCheckedException)) if (!(e instanceof ClusterTopologyCheckedException))
Expand All @@ -1629,7 +1640,7 @@ private class TopologyListener implements GridLocalEventListener {
} }


if (!retries.isEmpty()) if (!retries.isEmpty())
onReassignmentFailed(topVer.topologyVersion(), retries); onReassignmentFailed(topVer, retries);
} }


// Clean up zombie assignments. // Clean up zombie assignments.
Expand Down Expand Up @@ -1666,13 +1677,14 @@ private class TopologyListener implements GridLocalEventListener {
* @param topVer Topology version. * @param topVer Topology version.
* @param retries Retries. * @param retries Retries.
*/ */
private void onReassignmentFailed(final long topVer, final Collection<GridServiceDeployment> retries) { private void onReassignmentFailed(final AffinityTopologyVersion topVer,
final Collection<GridServiceDeployment> retries) {
if (!busyLock.enterBusy()) if (!busyLock.enterBusy())
return; return;


try { try {
// If topology changed again, let next event handle it. // If topology changed again, let next event handle it.
if (ctx.discovery().topologyVersion() != topVer) if (ctx.discovery().topologyVersionEx().equals(topVer))
return; return;


for (Iterator<GridServiceDeployment> it = retries.iterator(); it.hasNext(); ) { for (Iterator<GridServiceDeployment> it = retries.iterator(); it.hasNext(); ) {
Expand Down

0 comments on commit a62a013

Please sign in to comment.