no need to synchronize, since EventManager will execute sequentially
339303
public static void downloadDistributeStormCode(Map conf, String topologyId, String masterCodeDir)
            throws IOException, TException {
        String tmpToot = null;
        try {
            // STORM_LOCAL_DIR/supervisor/tmp/(UUID)
            tmpToot = StormConfig.supervisorTmpDir(conf) + File.separator + UUID.randomUUID().toString();

            // STORM_LOCAL_DIR/supervisor/stormdist/topologyId
            String stormRoot = StormConfig.supervisor_stormdist_root(conf, topologyId);

//        JStormServerUtils.downloadCodeFromMaster(conf, tmproot, masterCodeDir, topologyId, true);
            JStormServerUtils.downloadCodeFromBlobStore(conf, tmpToot, topologyId);

            // tmproot/stormjar.jar
            String localFileJarTmp = StormConfig.stormjar_path(tmpToot);

            // extract dir from jar
            JStormUtils.extractDirFromJar(localFileJarTmp, StormConfig.RESOURCES_SUBDIR, tmpToot);

            File srcDir = new File(tmpToot);
            File destDir = new File(stormRoot);
            try {
                FileUtils.moveDirectory(srcDir, destDir);
            } catch (FileExistsException e) {
                FileUtils.copyDirectory(srcDir, destDir);
                FileUtils.deleteQuietly(srcDir);
            }
        } finally {
            if (tmpToot != null) {
                File srcDir = new File(tmpToot);
                FileUtils.deleteQuietly(srcDir);
            }
        }
    }
769719
final void performAction(final AbstractEventQueue eventQueue, final Event<?> event) {
        final Event<T> eventObject = (Event<T>) event;

        try {
            action(eventObject);
        } catch (final Exception e) {
            LOGGER.log(Level.WARN, "Event perform failed", e);
        } finally { // remove event from event queue
            if (eventQueue instanceof SynchronizedEventQueue) {
                final SynchronizedEventQueue synchronizedEventQueue = (SynchronizedEventQueue) eventQueue;

                synchronizedEventQueue.removeEvent(eventObject);
            }
        }
    }
541984
public void async() {
		final Request request = data.build();
		final CallbackHandler handler = request.handler != null ? request.handler : singleton.handler;
		final boolean safe = request.safe != null ? request.safe : singleton.safe;

		// Ignore current request if already running (ignoreIfRunning)
		synchronized (activeRequests) {
			if (ignoreIfRunning && activeRequests.contains(request.tag)) {
				singleton.log(request.tag + " request already running! Ignoring...");
				return;
			}
			activeRequests.add(request.tag);
		}

		new AsyncTask<Void, Void, Void>() {
			@Override
			protected Void doInBackground(Void... params) {
				try {
					// Delay thread if specified
					if (delay > 0) {
						singleton.log("Delaying " + request.tag + " request for " + delay + "ms");
						Thread.sleep(delay);
					}

					// Execute request
					try {
						singleton.load(request);
					} catch (IOException e) {
						singleton.log("Error executing request " + request.tag + " asynchronously! " + e.getMessage(), Log.ERROR);
						handler.onLoadFailed(safe, request.callback.get());
					}
				} catch (InterruptedException e) {
					singleton.log(request.tag + " thread interrupted!");
				} finally {
					synchronized (activeRequests) {
						activeRequests.remove(request.tag);
					}
				}
				return null;
			}

			@Override
			protected void onCancelled(Void aVoid) {
				super.onCancelled(aVoid);

				// Double check to make sure this request is removed
				synchronized (activeRequests) {
					activeRequests.remove(request.tag);
				}
			}
		}.executeOnExecutor(AsyncTask.SERIAL_EXECUTOR);
	}
199362
function() {

		if (this._binds) {

			for (var i = 0, count = this._binds.length; i < count; i++) {
				this._binds[i].removeListener(this._bind_changed_listeners[i]);
			}

			this._binds = this._bind_changed_listeners = null;

		}

		this._argument.removeListener(this._argument_changed_listener);
		this._observable_listener && this._flushObservable();

		if (this._own_collection) {

			this._value.destroy();

		}

		this.suspendRefreshable();

	}
370263
public void synchronize(List<? extends E> list) {
		assert SwingUtilities.isEventDispatchThread() : ApplicationResources.accessor.getMessage("assert.notRunningInSwingEventThread");

		// Make sure the first element exists and matches
		int modelOffset;
		if(constantFirstRow!=null) {
			modelOffset = 1;
			if(getSize()==0) addElement(constantFirstRow);
			else if(!getElementAt(0).equals(constantFirstRow)) {
				insertElementAt(constantFirstRow, 0);
			}
		} else modelOffset = 0;

		// Synchronize the dynamic part of the list
		int size = list.size();
		for(int index=0; index<size; index++) {
			E obj = list.get(index);
			if(index>=(getSize()-modelOffset)) addElement(obj);
			else if(!obj.equals(getElementAt(index+modelOffset))) {
				// Objects don't match
				// If this object is found further down the list, then delete up to that object
				int foundIndex = -1;
				for(int searchIndex = index+1; searchIndex<(getSize()-modelOffset); searchIndex++) {
					if(obj.equals(getElementAt(searchIndex+modelOffset))) {
						foundIndex = searchIndex;
						break;
					}
				}
				if(foundIndex!=-1) {
					// No removeRange
					// removeRange(index+modelOffset, foundIndex-1+modelOffset);
					for(
						int removeIndex = foundIndex-1+modelOffset, end = index+modelOffset;
						removeIndex>=end;
						removeIndex--
					) removeElementAt(removeIndex);
				} else {
					// Otherwise, insert in the current index
					insertElementAt(obj, index+modelOffset);
				}
			}
		}
		// Remove any extra
		while((getSize()-modelOffset) > size) removeElementAt(getSize()-1);
	}
989021
void upsertEvent(EventId eventId, Lockable<V> event) {
		this.eventsBufferCache.put(eventId, event);
	}
948478
def asynchronous(function, event):
    
    thread = Thread(target=synchronous, args=(function, event))
    thread.daemon = True
    thread.start()
320209
public void dispatch_event(final EventData eventData) {
        final TangoDataReady tangoDataReady = this;
        if (EventUtil.graphicAvailable()) {
                //   Causes doRun.run() to be executed asynchronously
                //      on the AWT event dispatching thread.
            Runnable do_work_later = new Runnable() {
                public void run() {
                    fireTangoDataReadyEvent(tangoDataReady, eventData);
                }
            };
            SwingUtilities.invokeLater(do_work_later);
        }
        else
            fireTangoDataReadyEvent(tangoDataReady, eventData);
    }
1214051
@Override
  public void eventReceived(EventSubscriptionEntity eventSubscriptionEntity, Object payload, boolean processASync) {
    if (processASync) {
      scheduleEventAsync(eventSubscriptionEntity, payload);
    } else {
      processEventSync(eventSubscriptionEntity, payload);
    }
  }
1244046
@Override
    public void removeSynchronizationObserver(Observer<Registry<KEY, ENTRY>, Long> observer) {
        synchronisationObservable.removeObserver(observer);
    }
403723
private QYBaseRespMsg processEventHandle(QYBaseEvent event){
        if(CollectionUtil.isEmpty(eventHandles)){
            synchronized (LOCK){
                eventHandles = this.initEventHandles();
            }
        }
        if(CollectionUtil.isNotEmpty(eventHandles)){
            for(QYEventHandle eventHandle : eventHandles){
                QYBaseRespMsg resultMsg = null;
                boolean result;
                try{
                    result = eventHandle.beforeHandle(event);
                }catch (Exception e){
                    result = false;
                }
                if(result){
                    resultMsg = eventHandle.handle(event);
                }
                if(BeanUtil.nonNull(resultMsg)){
                    return resultMsg;
                }
            }
        }
        return null;
    }
338610
@Override
  public void onNext(final T event) {
    if (LOG.isLoggable(Level.FINE)) {
      LOG.log(Level.FINE, "remoteid: {0}\n{1}", new Object[]{remoteId.getSocketAddress(), event.toString()});
    }
    handler.onNext(new RemoteEvent<T>(myId.getSocketAddress(), remoteId.getSocketAddress(),
        seqGen.getNextSeq(remoteId.getSocketAddress()), event));
  }
419335
void storeNextEvent() {
    try {
      if (currentStream != null && currentStream.isOpen()) {
        final StitchEvent<ChangeEvent<BsonDocument>> event = currentStream.nextEvent();
        if (event == null) {
          return;
        }

        if (event.getError() != null) {
          throw event.getError();
        }

        if (event.getData() == null) {
          return;
        }

        logger.debug(String.format(Locale.US,
            "NamespaceChangeStreamListener::stream ns=%s event found: op=%s id=%s",
            nsConfig.getNamespace(), event.getData().getOperationType(), event.getData().getId()));
        nsLock.writeLock().lockInterruptibly();
        try {
          events.put(BsonUtils.getDocumentId(event.getData().getDocumentKey()), event.getData());
        } finally {
          nsLock.writeLock().unlock();
        }

        for (final Callback<ChangeEvent<BsonDocument>, Object> watcher : watchers) {
          watcher.onComplete(OperationResult.successfulResultOf(event.getData()));
        }
      }
    } catch (final InterruptedException | IOException ex) {
      logger.info(String.format(
          Locale.US,
          "NamespaceChangeStreamListener::stream ns=%s interrupted on "
              + "fetching next event: %s",
          nsConfig.getNamespace(),
          ex));
      logger.info("stream END – INTERRUPTED");
      Thread.currentThread().interrupt();
    } catch (final Exception ex) {
      // TODO: Emit error through DataSynchronizer as an ifc
      logger.error(String.format(
          Locale.US,
          "NamespaceChangeStreamListener::stream ns=%s exception on fetching next event: %s",
          nsConfig.getNamespace(),
          ex), ex);
      logger.info("stream END – EXCEPTION");
      final boolean wasInterrupted = Thread.currentThread().isInterrupted();
      this.close();
      if (wasInterrupted) {
        Thread.currentThread().interrupt();
      }
    }
  }
188255
public function dispatchAsyncListenersOnly($eventName, Event $event = null)
    {
        try {
            $this->asyncListener->syncMode($eventName);

            parent::dispatch($eventName, $event);
        } finally {
            $this->asyncListener->resetSyncMode();
        }
    }
739730
public void finallyDo(Callback resultCB) {
        // FIXME: this can be implemented more efficient
        while( !lock.compareAndSet(false,true) ) {}
        try {
            if (resultReceiver != null)
                throw new RuntimeException("Double register of future listener");
            resultReceiver = resultCB;
            if (hadResult) {
                hasFired = true;
                lock.set(false);
                resultCB.complete(result, error);
            }
        } finally {
            lock.set(false);
        }
    }
1083785
private void getNextEventIndex() {
        for (int i = next + 1; i < events.size(); i++) {
            String evName = getValueOr(events.get(i), "event", "");
            if (evName.equals(eventType)) {
                next = i;
                return;
            }
        }
        next = events.size();
    }
1066253
private <T, S extends ISyncableData> void doSync(T caller, String... syncNames)
	{
		@SuppressWarnings("unchecked")
		ISyncHandler<T, S> handler = (ISyncHandler<T, S>) getHandler(caller);
		if (handler == null)
			return;

		S data = handler.getSyncData(caller);
		int indexes = getFieldIndexes(handler, syncNames);
		Map<String, Object> values = getFieldValues(caller, handler, syncNames);

		SyncerMessage.Packet<T, S> packet = new Packet<>(getHandlerId(caller.getClass()), data, indexes, values);

		handler.send(caller, packet);
	}