| /* |
| * Licensed to the Apache Software Foundation (ASF) under one or more contributor license |
| * agreements. See the NOTICE file distributed with this work for additional information regarding |
| * copyright ownership. The ASF licenses this file to You under the Apache License, Version 2.0 (the |
| * "License"); you may not use this file except in compliance with the License. You may obtain a |
| * copy of the License at |
| * |
| * http://www.apache.org/licenses/LICENSE-2.0 |
| * |
| * Unless required by applicable law or agreed to in writing, software distributed under the License |
| * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express |
| * or implied. See the License for the specific language governing permissions and limitations under |
| * the License. |
| */ |
| package org.apache.geode.internal.cache.wan.serial; |
| |
| import java.io.IOException; |
| import java.util.Collections; |
| import java.util.Iterator; |
| import java.util.LinkedHashMap; |
| import java.util.Map; |
| import java.util.Random; |
| import java.util.Set; |
| import java.util.concurrent.Executor; |
| import java.util.concurrent.ExecutorService; |
| import java.util.stream.Collectors; |
| |
| import org.apache.logging.log4j.Logger; |
| |
| import org.apache.geode.CancelException; |
| import org.apache.geode.SystemFailure; |
| import org.apache.geode.cache.CacheException; |
| import org.apache.geode.cache.CacheListener; |
| import org.apache.geode.cache.EntryEvent; |
| import org.apache.geode.cache.Operation; |
| import org.apache.geode.cache.Region; |
| import org.apache.geode.cache.RegionDestroyedException; |
| import org.apache.geode.cache.wan.GatewayQueueEvent; |
| import org.apache.geode.cache.wan.GatewaySender; |
| import org.apache.geode.distributed.DistributedSystem; |
| import org.apache.geode.internal.Assert; |
| import org.apache.geode.internal.cache.DistributedRegion; |
| import org.apache.geode.internal.cache.EntryEventImpl; |
| import org.apache.geode.internal.cache.EnumListenerEvent; |
| import org.apache.geode.internal.cache.EventID; |
| import org.apache.geode.internal.cache.LocalRegion; |
| import org.apache.geode.internal.cache.wan.AbstractGatewaySender; |
| import org.apache.geode.internal.cache.wan.AbstractGatewaySender.EventWrapper; |
| import org.apache.geode.internal.cache.wan.AbstractGatewaySenderEventProcessor; |
| import org.apache.geode.internal.cache.wan.GatewaySenderEventCallbackArgument; |
| import org.apache.geode.internal.cache.wan.GatewaySenderEventCallbackDispatcher; |
| import org.apache.geode.internal.cache.wan.GatewaySenderEventImpl; |
| import org.apache.geode.internal.cache.wan.GatewaySenderStats; |
| import org.apache.geode.internal.logging.LogService; |
| import org.apache.geode.internal.logging.LoggingExecutors; |
| import org.apache.geode.internal.monitoring.ThreadsMonitoring; |
| import org.apache.geode.pdx.internal.PeerTypeRegistration; |
| |
| /** |
| * @since GemFire 7.0 |
| * |
| */ |
| public class SerialGatewaySenderEventProcessor extends AbstractGatewaySenderEventProcessor { |
| private static final Logger logger = LogService.getLogger(); |
| |
| private final Object unprocessedEventsLock = new Object(); |
| |
| protected static final int RANDOM_SLEEP_TIME = 1000; |
| /** |
| * A <code>Map</code> of events that have not been processed by the primary yet. This map is |
| * created and used by a secondary <code>GatewaySender</code> to keep track of events that have |
| * been received by the secondary but not yet processed by the primary. Once an event has been |
| * processed by the primary, it is removed from this map. This map will only be used in the event |
| * that this <code>GatewaySender</code> becomes the primary. Any events contained in this map will |
| * need to be sent to other <code>GatewayReceiver</code>s. Note: unprocessedEventsLock MUST be |
| * synchronized before using this map. |
| */ |
| private Map<EventID, EventWrapper> unprocessedEvents; |
| |
| /** |
| * A <code>Map</code> of tokens (i.e. longs) of entries that we have heard of from the primary but |
| * not yet the secondary. This map is created and used by a secondary <code>GatewaySender</code> |
| * to keep track. Note: unprocessedEventsLock MUST be synchronized before using this map. This is |
| * not a cut and paste error. sync unprocessedEventsLock when using unprocessedTokens. |
| */ |
| protected Map<EventID, Long> unprocessedTokens; |
| |
| private ExecutorService executor; |
| |
| private Object listenerObjectLock = new Object(); |
| |
| private boolean failoverCompleted = false; |
| |
| private final Object failoverCompletedLock = new Object(); |
| |
| /** |
| * When the Number of unchecked events exceeds this threshold and the number of tokens in the map |
| * exceeds this threshold then a check will be done for old tokens. |
| */ |
| protected static final int REAP_THRESHOLD = 1000; |
| |
| /* |
| * How many events have happened without a reap check being done? |
| */ |
| private int uncheckedCount = 0; |
| |
| |
| public SerialGatewaySenderEventProcessor(AbstractGatewaySender sender, String id, |
| ThreadsMonitoring tMonitoring) { |
| super("Event Processor for GatewaySender_" + id, sender, tMonitoring); |
| |
| this.unprocessedEvents = new LinkedHashMap<EventID, EventWrapper>(); |
| this.unprocessedTokens = new LinkedHashMap<EventID, Long>(); |
| |
| initializeMessageQueue(id); |
| } |
| |
| @Override |
| protected void initializeMessageQueue(String id) { |
| // Create the region name |
| StringBuffer regionNameBuffer = new StringBuffer(); |
| regionNameBuffer.append(id).append("_SERIAL_GATEWAY_SENDER_QUEUE"); |
| String regionName = regionNameBuffer.toString(); |
| |
| CacheListener listener = null; |
| if (!this.sender.isPrimary()) { |
| listener = new SerialSecondaryGatewayListener(this); |
| initializeListenerExecutor(); |
| } |
| // Create the region queue |
| this.queue = new SerialGatewaySenderQueue(sender, regionName, listener); |
| |
| if (logger.isDebugEnabled()) { |
| logger.debug("Created queue: {}", this.queue); |
| } |
| } |
| |
| /** |
| * @return false on failure |
| */ |
| protected boolean waitForPrimary() { |
| |
| try { |
| this.sender.getSenderAdvisor().waitToBecomePrimary(this); |
| } catch (InterruptedException e) { |
| // No need to set the interrupt bit, we're exiting the thread. |
| if (!stopped()) { |
| logger.fatal( |
| "An InterruptedException occurred. The thread will exit.", |
| e); |
| } |
| shutdownListenerExecutor(); |
| return false; |
| } |
| try { |
| shutdownListenerExecutor(); |
| |
| // Don't allow conserve-sockets = false so that ordering is preserved. |
| DistributedSystem.setThreadsSocketPolicy(true); |
| |
| // Once notification has occurred, handle failover |
| if (!stopped()) { |
| handleFailover(); |
| } else { |
| return false; |
| } |
| } catch (RegionDestroyedException e) { |
| // This happens during handleFailover |
| // because the region on _eventQueue can be closed. |
| if (!stopped()) { |
| logger.debug("Terminating due to {}", e.getMessage(), e); |
| } |
| return false; |
| } catch (CancelException e) { |
| if (!stopped()) { |
| logger.debug("Terminating due to {}", e.getMessage(), e); |
| } |
| return false; |
| } finally { |
| // This is a good thing even on termination, because it ends waits |
| // on the part of other threads. |
| completeFailover(); |
| } |
| return true; |
| } |
| |
| |
| |
| @Override |
| public void run() { |
| try { |
| setRunningStatus(); |
| // If this is not a primary gateway, wait for notification |
| if (!sender.isPrimary()) { |
| if (!waitForPrimary()) { |
| return; |
| } |
| } else { |
| // we are the primary so mark failover as being completed |
| completeFailover(); |
| } |
| // Begin to process the message queue after becoming primary |
| if (logger.isDebugEnabled()) { |
| logger.debug("Beginning to process the message queue"); |
| } |
| |
| if (!sender.isPrimary()) { |
| logger.warn("About to process the message queue but not the primary."); |
| } |
| |
| // Sleep for a bit. The random is so that if several of these are |
| // started at once, they will stagger a bit. |
| try { |
| Thread.sleep(new Random().nextInt(RANDOM_SLEEP_TIME)); |
| } catch (InterruptedException e) { |
| // no need to set the interrupt bit or throw an exception; just exit. |
| return; |
| } |
| processQueue(); |
| } catch (CancelException e) { |
| if (!this.isStopped()) { |
| logger.info("A cancellation occurred. Stopping the dispatcher."); |
| setIsStopped(true); |
| } |
| } catch (VirtualMachineError err) { |
| SystemFailure.initiateFailure(err); |
| // If this ever returns, rethrow the error. We're poisoned |
| // now, so don't let this thread continue. |
| throw err; |
| } catch (Throwable e) { |
| // Whenever you catch Error or Throwable, you must also |
| // catch VirtualMachineError (see above). However, there is |
| // _still_ a possibility that you are dealing with a cascading |
| // error condition, so you also need to check to see if the JVM |
| // is still usable: |
| SystemFailure.checkFailure(); |
| logger.fatal( |
| "Message dispatch failed due to unexpected exception..", e); |
| } |
| } |
| |
| @Override |
| protected void rebalance() { |
| // No reason to rebalance a serial sender since all connections are to the same server. |
| throw new UnsupportedOperationException(); |
| } |
| |
| /** |
| * Handle failover. This method is called when a secondary <code>GatewaySender</code> becomes a |
| * primary <code>GatewaySender</code>. |
| * |
| * Once this secondary becomes the primary, it must: |
| * <ul> |
| * <li>Remove the queue's CacheListener |
| * <li>Process the map of unprocessed events (those it has seen but the previous primary had not |
| * yet processed before it crashed). These will include both queued and unqueued events. Remove |
| * from the queue any events that were already sent |
| * <li>Clear the unprocessed events map |
| * </ul> |
| */ |
| protected void handleFailover() { |
| /* |
| * We must hold this lock while we're processing these maps to prevent us from handling a |
| * secondary event while failover occurs. See enqueueEvent |
| */ |
| synchronized (this.unprocessedEventsLock) { |
| // Remove the queue's CacheListener |
| this.queue.removeCacheListener(); |
| this.unprocessedTokens = null; |
| |
| // Process the map of unprocessed events |
| logger.info("Gateway Failover Initiated: Processing {} unprocessed events.", |
| this.unprocessedEvents.size()); |
| GatewaySenderStats statistics = this.sender.getStatistics(); |
| if (!this.unprocessedEvents.isEmpty()) { |
| // do a reap for bug 37603 |
| reapOld(statistics, true); // to get rid of timed out events |
| // now iterate over the region queue to figure out what unprocessed |
| // events are already in the queue |
| { |
| Iterator it = this.queue.getRegion().values().iterator(); |
| while (it.hasNext() && !stopped()) { |
| Object o = it.next(); |
| if (o != null && o instanceof GatewaySenderEventImpl) { |
| GatewaySenderEventImpl ge = (GatewaySenderEventImpl) o; |
| EventWrapper unprocessedEvent = this.unprocessedEvents.remove(ge.getEventId()); |
| if (unprocessedEvent != null) { |
| unprocessedEvent.event.release(); |
| if (this.unprocessedEvents.isEmpty()) { |
| break; |
| } |
| } |
| } |
| } |
| } |
| // now for every unprocessed event add it to the end of the queue |
| { |
| Iterator<Map.Entry<EventID, EventWrapper>> it = |
| this.unprocessedEvents.entrySet().iterator(); |
| while (it.hasNext()) { |
| if (stopped()) |
| break; |
| Map.Entry<EventID, EventWrapper> me = it.next(); |
| EventWrapper ew = me.getValue(); |
| GatewaySenderEventImpl gatewayEvent = ew.event; |
| // Initialize each gateway event. This initializes the key, |
| // value |
| // and callback arg based on the EntryEvent. |
| // TODO:wan70, remove dependencies from old code |
| gatewayEvent.initialize(); |
| // Verify that they GatewayEventCallbackArgument is initialized. |
| // If not, initialize it. It won't be initialized if a client to |
| // this GatewayHub VM was the creator of this event. This Gateway |
| // will be the first one to process it. If will be initialized if |
| // this event was sent to this Gateway from another GatewayHub |
| // (either directly or indirectly). |
| GatewaySenderEventCallbackArgument seca = gatewayEvent.getSenderCallbackArgument(); |
| if (seca.getOriginatingDSId() == GatewaySender.DEFAULT_DISTRIBUTED_SYSTEM_ID) { |
| seca.setOriginatingDSId(sender.getMyDSId()); |
| seca.initializeReceipientDSIds(Collections.singletonList(sender.getRemoteDSId())); |
| } |
| it.remove(); |
| boolean queuedEvent = false; |
| try { |
| queuedEvent = queuePrimaryEvent(gatewayEvent); |
| } catch (IOException ex) { |
| if (!stopped()) { |
| logger.warn( |
| String.format("Event dropped during failover: %s", gatewayEvent), |
| ex); |
| } |
| } catch (CacheException ex) { |
| if (!stopped()) { |
| logger.warn( |
| String.format("Event dropped during failover: %s", gatewayEvent), |
| ex); |
| } |
| } finally { |
| if (!queuedEvent) { |
| gatewayEvent.release(); |
| } |
| } |
| } |
| } |
| // Clear the unprocessed events map |
| statistics.clearUnprocessedMaps(); |
| } |
| |
| // Iterate the entire queue and mark all events as possible |
| // duplicate |
| logger.info("{} : Marking {} events as possible duplicates", |
| getSender(), Integer.valueOf(this.queue.size())); |
| Iterator it = this.queue.getRegion().values().iterator(); |
| while (it.hasNext() && !stopped()) { |
| Object o = it.next(); |
| if (o != null && o instanceof GatewaySenderEventImpl) { |
| GatewaySenderEventImpl ge = (GatewaySenderEventImpl) o; |
| ge.setPossibleDuplicate(true); |
| } |
| } |
| |
| releaseUnprocessedEvents(); |
| } // synchronized |
| } |
| |
| private void releaseUnprocessedEvents() { |
| synchronized (this.unprocessedEventsLock) { |
| Map<EventID, EventWrapper> m = this.unprocessedEvents; |
| if (m != null) { |
| for (EventWrapper ew : m.values()) { |
| GatewaySenderEventImpl gatewayEvent = ew.event; |
| if (logger.isDebugEnabled()) { |
| logger.debug("releaseUnprocessedEvents:" + gatewayEvent); |
| } |
| gatewayEvent.release(); |
| } |
| this.unprocessedEvents = null; |
| } |
| } |
| } |
| |
| @Override |
| public void closeProcessor() { |
| try { |
| super.closeProcessor(); |
| } finally { |
| releaseUnprocessedEvents(); |
| } |
| } |
| |
| /** |
| * Add the input object to the event queue |
| */ |
| @Override |
| public void enqueueEvent(EnumListenerEvent operation, EntryEvent event, Object substituteValue) |
| throws IOException, CacheException { |
| // There is a case where the event is serialized for processing. The |
| // region is not |
| // serialized along with the event since it is a transient field. I |
| // created an |
| // intermediate object (GatewayEventImpl) to avoid this since the region |
| // name is |
| // used in the sendBatch method, and it can't be null. See EntryEventImpl |
| // for details. |
| GatewaySenderEventImpl senderEvent = null; |
| |
| boolean isPrimary = sender.isPrimary(); |
| if (!isPrimary) { |
| // Fix for #40615. We need to check if we've now become the primary |
| // while holding the unprocessedEventsLock. This prevents us from failing |
| // over while we're processing an event as a secondaryEvent. |
| synchronized (unprocessedEventsLock) { |
| // Test whether this gateway is the primary. |
| if (sender.isPrimary()) { |
| isPrimary = true; |
| } else { |
| // If it is not, create an uninitialized GatewayEventImpl and |
| // put it into the map of unprocessed events, except 2 Special cases: |
| // 1) UPDATE_VERSION_STAMP: only enqueue to primary |
| // 2) CME && !originRemote: only enqueue to primary |
| boolean isUpdateVersionStamp = |
| event.getOperation().equals(Operation.UPDATE_VERSION_STAMP); |
| boolean isCME_And_NotOriginRemote = |
| ((EntryEventImpl) event).isConcurrencyConflict() && !event.isOriginRemote(); |
| if (!(isUpdateVersionStamp || isCME_And_NotOriginRemote)) { |
| senderEvent = |
| new GatewaySenderEventImpl(operation, event, substituteValue, false); |
| handleSecondaryEvent(senderEvent); |
| } |
| } |
| } |
| } |
| if (isPrimary) { |
| Region region = event.getRegion(); |
| boolean isPDXRegion = (region instanceof DistributedRegion |
| && region.getName().equals(PeerTypeRegistration.REGION_NAME)); |
| if (!isPDXRegion) { |
| waitForFailoverCompletion(); |
| } |
| // If it is, create and enqueue an initialized GatewayEventImpl |
| senderEvent = new GatewaySenderEventImpl(operation, event, substituteValue); // OFFHEAP ok |
| |
| boolean queuedEvent = false; |
| try { |
| queuedEvent = queuePrimaryEvent(senderEvent); |
| } finally { |
| // When queuePrimaryEvent() failed with some exception, it could |
| // occur after the GatewaySenderEventImpl is put onto the queue. |
| // In that case, the GatewaySenderEventImpl could be released here, |
| // and IllegalStateException could be thrown if getDeserializedValue is called |
| // when the event is accessed through the region queue. |
| if (!queuedEvent) { |
| GatewaySenderEventImpl.release(senderEvent); |
| } |
| } |
| } |
| } |
| |
| private boolean queuePrimaryEvent(GatewaySenderEventImpl gatewayEvent) |
| throws IOException, CacheException { |
| // Queue the event |
| GatewaySenderStats statistics = this.sender.getStatistics(); |
| if (logger.isDebugEnabled()) { |
| logger.debug("{}: Queueing event ({}): {}", sender.getId(), |
| (statistics.getEventsQueued() + 1), gatewayEvent); |
| } |
| if (!sender.beforeEnqueue(gatewayEvent)) { |
| if (logger.isDebugEnabled()) { |
| logger.debug("Event {} is not added to queue.", gatewayEvent); |
| } |
| statistics.incEventsFiltered(); |
| return false; |
| } |
| long start = statistics.startTime(); |
| boolean putDone = false; |
| try { |
| putDone = this.queue.put(gatewayEvent); |
| } catch (InterruptedException e) { |
| // Asif Not expected from SingleWriteSingleReadRegionQueue as it does not |
| // throw |
| // InterruptedException. But since both HARegionQueue and |
| // SingleReadSingleWriteRegionQueue |
| // extend RegionQueue , it has to handle InterruptedException |
| Thread.currentThread().interrupt(); |
| getSender().getCancelCriterion().checkCancelInProgress(e); |
| } |
| statistics.endPut(start); |
| if (logger.isDebugEnabled()) { |
| logger.debug("{}: Queued event ({}): {}", sender.getId(), (statistics.getEventsQueued()), |
| gatewayEvent); |
| } |
| // this._logger.warning(getGateway() + ": Queued event (" + |
| // (statistics.getEventsQueued()) + "): " + gatewayEvent + " queue size: " |
| // + this._eventQueue.size()); |
| /* |
| * FAILOVER TESTING CODE System.out.println(getName() + ": Queued event (" + |
| * (statistics.getEventsQueued()) + "): " + gatewayEvent.getId()); |
| */ |
| int queueSize = eventQueueSize(); |
| statistics.incQueueSize(1); |
| if (!this.eventQueueSizeWarning && queueSize >= AbstractGatewaySender.QUEUE_SIZE_THRESHOLD) { |
| logger.warn("{}: The event queue has reached {} events. Processing will continue.", |
| sender.getId(), |
| Integer.valueOf(AbstractGatewaySender.QUEUE_SIZE_THRESHOLD)); |
| this.eventQueueSizeWarning = true; |
| } |
| return putDone; |
| } |
| |
| protected void waitForFailoverCompletion() { |
| synchronized (this.failoverCompletedLock) { |
| if (this.failoverCompleted) { |
| return; |
| } |
| logger.info("{} : Waiting for failover completion", this); |
| try { |
| while (!this.failoverCompleted) { |
| this.failoverCompletedLock.wait(); |
| } |
| } catch (InterruptedException ex) { |
| Thread.currentThread().interrupt(); |
| this.sender.getCache().getCancelCriterion().checkCancelInProgress(ex); |
| logger.info("{}: did not wait for failover completion due to interruption.", |
| this); |
| } |
| } |
| } |
| |
| protected void completeFailover() { |
| synchronized (this.failoverCompletedLock) { |
| this.failoverCompleted = true; |
| this.failoverCompletedLock.notifyAll(); |
| } |
| } |
| |
| /** |
| * Update an unprocessed event in the unprocessed events map. This method is called by a secondary |
| * <code>GatewaySender</code> to store a gateway event until it is processed by a primary |
| * <code>GatewaySender</code>. The complexity of this method is the fact that the event could be |
| * processed first by either the primary or secondary <code>GatewaySender</code>. |
| * |
| * If the primary processes the event first, the map will already contain an entry for the event |
| * (through |
| * {@link org.apache.geode.internal.cache.wan.serial.SerialSecondaryGatewayListener#afterDestroy} |
| * ). When the secondary processes the event, it will remove it from the map. |
| * |
| * If the secondary processes the event first, it will add it to the map. When the primary |
| * processes the event (through |
| * {@link org.apache.geode.internal.cache.wan.serial.SerialSecondaryGatewayListener#afterDestroy} |
| * ), it will then be removed from the map. |
| * |
| * @param senderEvent The event being processed |
| */ |
| protected void handleSecondaryEvent(GatewaySenderEventImpl senderEvent) { |
| basicHandleSecondaryEvent(senderEvent); |
| } |
| |
| /** |
| * Update an unprocessed event in the unprocessed events map. This method is called by a primary |
| * <code>Gateway</code> (through |
| * {@link org.apache.geode.internal.cache.wan.serial.SerialSecondaryGatewayListener#afterCreate} ) |
| * to notify the secondary <code>Gateway</code> that an event has been added to the queue. Once an |
| * event has been added to the queue, the secondary no longer needs to keep track of it in the |
| * unprocessed events map. The complexity of this method is the fact that the event could be |
| * processed first by either the primary or secondary <code>Gateway</code>. |
| * |
| * If the primary processes the event first, the map will not contain an entry for the event. It |
| * will be added to the map in this case so that when the secondary processes it, it will know |
| * that the primary has already processed it, and it can be safely removed. |
| * |
| * If the secondary processes the event first, the map will already contain an entry for the |
| * event. In this case, the event can be removed from the map. |
| * |
| * @param gatewayEvent The event being processed |
| */ |
| protected void handlePrimaryEvent(final GatewaySenderEventImpl gatewayEvent) { |
| Executor my_executor = this.executor; |
| synchronized (listenerObjectLock) { |
| if (my_executor == null) { |
| // should mean we are now primary |
| return; |
| } |
| my_executor.execute(new Runnable() { |
| @Override |
| public void run() { |
| basicHandlePrimaryEvent(gatewayEvent); |
| } |
| }); |
| } |
| } |
| |
| /** |
| * Called when the primary gets rid of an event from the queue This method added to fix bug 37603 |
| */ |
| protected void handlePrimaryDestroy(final GatewaySenderEventImpl gatewayEvent) { |
| Executor my_executor = this.executor; |
| synchronized (listenerObjectLock) { |
| if (my_executor == null) { |
| // should mean we are now primary |
| return; |
| } |
| my_executor.execute(new Runnable() { |
| @Override |
| public void run() { |
| basicHandlePrimaryDestroy(gatewayEvent.getEventId()); |
| } |
| }); |
| } |
| } |
| |
| /** |
| * Just remove the event from the unprocessed events map if it is present. This method added to |
| * fix bug 37603 |
| */ |
| protected boolean basicHandlePrimaryDestroy(final EventID eventId) { |
| if (this.sender.isPrimary()) { |
| // no need to do anything if we have become the primary |
| return false; |
| } |
| GatewaySenderStats statistics = this.sender.getStatistics(); |
| // Get the event from the map |
| synchronized (unprocessedEventsLock) { |
| if (this.unprocessedEvents == null) |
| return false; |
| // now we can safely use the unprocessedEvents field |
| EventWrapper ew = this.unprocessedEvents.remove(eventId); |
| if (ew != null) { |
| ew.event.release(); |
| statistics.incUnprocessedEventsRemovedByPrimary(); |
| return true; |
| } |
| } |
| return false; |
| } |
| |
| protected void basicHandlePrimaryEvent(final GatewaySenderEventImpl gatewayEvent) { |
| if (this.sender.isPrimary()) { |
| // no need to do anything if we have become the primary |
| return; |
| } |
| GatewaySenderStats statistics = this.sender.getStatistics(); |
| // Get the event from the map |
| synchronized (unprocessedEventsLock) { |
| if (this.unprocessedEvents == null) |
| return; |
| // now we can safely use the unprocessedEvents field |
| EventWrapper ew = this.unprocessedEvents.remove(gatewayEvent.getEventId()); |
| |
| if (ew == null) { |
| // first time for the event |
| if (logger.isTraceEnabled()) { |
| logger.trace("{}: fromPrimary event {} : {}->{} added to unprocessed token map", |
| sender.getId(), gatewayEvent.getEventId(), gatewayEvent.getKey(), |
| gatewayEvent.getValueAsString(true)); |
| } |
| { |
| Long mapValue = |
| Long.valueOf(System.currentTimeMillis() + AbstractGatewaySender.TOKEN_TIMEOUT); |
| Long oldv = this.unprocessedTokens.put(gatewayEvent.getEventId(), mapValue); |
| if (oldv == null) { |
| statistics.incUnprocessedTokensAddedByPrimary(); |
| } else { |
| // its ok for oldv to be non-null |
| // this shouldn't happen anymore @todo add an assertion here |
| } |
| } |
| } else { |
| // already added by secondary (i.e. hub) |
| // the secondary |
| // gateway has already seen this event, and it can be safely |
| // removed (it already was above) |
| if (logger.isTraceEnabled()) { |
| logger.trace( |
| "{}: Primary create/update event {}:{}->{} remove from unprocessed events map", |
| sender.getId(), gatewayEvent.getEventId(), gatewayEvent.getKey(), |
| gatewayEvent.getValueAsString(true)); |
| } |
| ew.event.release(); |
| statistics.incUnprocessedEventsRemovedByPrimary(); |
| } |
| reapOld(statistics, false); |
| } |
| } |
| |
| private void basicHandleSecondaryEvent(final GatewaySenderEventImpl gatewayEvent) { |
| boolean freeGatewayEvent = true; |
| try { |
| GatewaySenderStats statistics = this.sender.getStatistics(); |
| // Get the event from the map |
| |
| if (!getSender().getGatewayEventFilters().isEmpty()) { |
| try { |
| gatewayEvent.initialize(); |
| } catch (Exception e) { |
| logger.warn( |
| String.format("Event failed to be initialized: %s", gatewayEvent), e); |
| } |
| if (!sender.beforeEnqueue(gatewayEvent)) { |
| statistics.incEventsFiltered(); |
| return; |
| } |
| } |
| Assert.assertHoldsLock(unprocessedEventsLock, true); |
| Assert.assertTrue(unprocessedEvents != null); |
| // @todo add an assertion that !getPrimary() |
| // now we can safely use the unprocessedEvents field |
| Long v = this.unprocessedTokens.remove(gatewayEvent.getEventId()); |
| |
| if (v == null) { |
| // first time for the event |
| if (logger.isTraceEnabled()) { |
| logger.trace("{}: fromSecondary event {}:{}->{} added from unprocessed events map", |
| sender.getId(), gatewayEvent.getEventId(), gatewayEvent.getKey(), |
| gatewayEvent.getValueAsString(true)); |
| } |
| { |
| EventWrapper mapValue = new EventWrapper(gatewayEvent); |
| EventWrapper oldv = this.unprocessedEvents.put(gatewayEvent.getEventId(), mapValue); |
| if (oldv == null) { |
| freeGatewayEvent = false; |
| statistics.incUnprocessedEventsAddedBySecondary(); |
| } else { |
| // put old one back in |
| this.unprocessedEvents.put(gatewayEvent.getEventId(), oldv); |
| // already added by secondary (i.e. hub) |
| logger.warn( |
| "{}: The secondary map already contained an event from hub {} so ignoring new event {}.", |
| sender.getId(), v, gatewayEvent); |
| } |
| } |
| } else { |
| // token already added by primary already removed |
| if (logger.isTraceEnabled()) { |
| logger.trace("{}: Secondary created event {}:{}->{} removed from unprocessed events map", |
| sender.getId(), gatewayEvent.getEventId(), gatewayEvent.getKey(), |
| gatewayEvent.getValueAsString(true)); |
| } |
| statistics.incUnprocessedTokensRemovedBySecondary(); |
| } |
| reapOld(statistics, false); |
| } finally { |
| if (freeGatewayEvent) { |
| gatewayEvent.release(); |
| } |
| } |
| } |
| |
| /** |
| * Call to check if a cleanup of tokens needs to be done |
| */ |
| private void reapOld(final GatewaySenderStats statistics, boolean forceEventReap) { |
| synchronized (this.unprocessedEventsLock) { |
| if (uncheckedCount > REAP_THRESHOLD) { // only check every X events |
| uncheckedCount = 0; |
| long now = System.currentTimeMillis(); |
| if (!forceEventReap && this.unprocessedTokens.size() > REAP_THRESHOLD) { |
| Iterator<Map.Entry<EventID, Long>> it = this.unprocessedTokens.entrySet().iterator(); |
| int count = 0; |
| while (it.hasNext()) { |
| Map.Entry<EventID, Long> me = it.next(); |
| long meValue = me.getValue().longValue(); |
| if (meValue <= now) { |
| // @todo log fine level message here |
| // it has expired so remove it |
| it.remove(); |
| count++; |
| } else { |
| // all done try again |
| break; |
| } |
| } |
| if (count > 0) { |
| // statistics.incUnprocessedTokensRemovedByTimeout(count); |
| } |
| } |
| if (forceEventReap || this.unprocessedEvents.size() > REAP_THRESHOLD) { |
| Iterator<Map.Entry<EventID, EventWrapper>> it = |
| this.unprocessedEvents.entrySet().iterator(); |
| int count = 0; |
| while (it.hasNext()) { |
| Map.Entry<EventID, EventWrapper> me = it.next(); |
| EventWrapper ew = me.getValue(); |
| if (ew.timeout <= now) { |
| // @todo log fine level message here |
| // it has expired so remove it |
| it.remove(); |
| ew.event.release(); |
| count++; |
| } else { |
| // all done try again |
| break; |
| } |
| } |
| if (count > 0) { |
| // statistics.incUnprocessedEventsRemovedByTimeout(count); |
| } |
| } |
| } else { |
| uncheckedCount++; |
| } |
| } |
| } |
| |
| @Override |
| public String toString() { |
| StringBuffer buffer = new StringBuffer(); |
| buffer.append("GatewayEventProcessor[").append("gatewaySenderId=").append(sender.getId()) |
| .append(";remoteDSId=").append(getSender().getRemoteDSId()).append(";batchSize=") |
| .append(getSender().getBatchSize()); |
| buffer.append("]"); |
| return buffer.toString(); |
| } |
| |
| /** |
| * Initialize the Executor that handles listener events. Only used by non-primary gateway senders |
| */ |
| private void initializeListenerExecutor() { |
| this.executor = |
| LoggingExecutors.newFixedThreadPoolWithTimeout("Queued Gateway Listener Thread", 1, 120); |
| } |
| |
| private void shutdownListenerExecutor() { |
| synchronized (listenerObjectLock) { |
| if (this.executor != null) { |
| this.executor.shutdown(); |
| this.executor = null; |
| } |
| } |
| } |
| |
| @Override |
| public void removeCacheListener() { |
| this.queue.removeCacheListener(); |
| } |
| |
| @Override |
| public void initializeEventDispatcher() { |
| if (logger.isDebugEnabled()) { |
| logger.debug(" Creating the GatewayEventCallbackDispatcher"); |
| } |
| this.dispatcher = new GatewaySenderEventCallbackDispatcher(this); |
| |
| } |
| |
| @Override |
| protected void enqueueEvent(GatewayQueueEvent event) { |
| // @TODO This API hasn't been implemented yet |
| throw new UnsupportedOperationException(); |
| } |
| |
| public void sendBatchDestroyOperationForDroppedEvent(EntryEventImpl dropEvent, int index) { |
| EntryEventImpl destroyEvent = |
| EntryEventImpl.create((LocalRegion) this.queue.getRegion(), Operation.DESTROY, (long) index, |
| null/* newValue */, null, false, sender.getCache().getMyId()); |
| destroyEvent.setEventId(dropEvent.getEventId()); |
| destroyEvent.disallowOffHeapValues(); |
| destroyEvent.setTailKey(-1L); |
| if (logger.isDebugEnabled()) { |
| logger.debug( |
| "SerialGatewaySenderEventProcessor sends BatchDestroyOperation to secondary for event {}", |
| destroyEvent); |
| } |
| |
| try { |
| BatchDestroyOperation op = new BatchDestroyOperation(destroyEvent); |
| op.distribute(); |
| if (logger.isDebugEnabled()) { |
| logger.debug("BatchRemovalThread completed destroy of dropped event {}", dropEvent); |
| } |
| } catch (Exception ignore) { |
| if (logger.isDebugEnabled()) { |
| logger.debug( |
| "Exception in sending dropped event could be ignored in order not to interrupt sender starting", |
| ignore); |
| } |
| } |
| } |
| |
| @Override |
| protected void registerEventDroppedInPrimaryQueue(EntryEventImpl droppedEvent) { |
| this.getSender().setModifiedEventId(droppedEvent); |
| sendBatchDestroyOperationForDroppedEvent(droppedEvent, -1); |
| } |
| |
| private String printEventIdList(Set<EventID> eventIds) { |
| StringBuffer sb = new StringBuffer().append("[").append( |
| eventIds.stream().map(entry -> entry.expensiveToString()).collect(Collectors.joining(", "))) |
| .append("]"); |
| return sb.toString(); |
| } |
| |
| @Override |
| public String printUnprocessedEvents() { |
| synchronized (this.unprocessedEventsLock) { |
| return printEventIdList(this.unprocessedEvents.keySet()); |
| } |
| } |
| |
| @Override |
| public String printUnprocessedTokens() { |
| synchronized (this.unprocessedEventsLock) { |
| return printEventIdList(this.unprocessedTokens.keySet()); |
| } |
| } |
| } |
