use synchronized blocks to protect pendingData.
511188
@Override
    public void sendData(TreeSet<Data> data) {
        if (data == null) {
            throw new IllegalArgumentException("Data must be not null");
        }
        if (data.isEmpty()) {
            return;
        }

        addData(data);

        if (distributed) {
            partitionManager.notifyData(new ArrayList<>(data));
        }
    }
786374
protected function pending()
    {
        $this->setData();
        foreach ($this->pending as $func => $args)
        {
            if (!empty($args))
            {
                call_user_func(array($this, $func . 'Pending'));
            }
        }

        //clear pending values after executed query
        $this->clearQuery();
    }
1331496
private void setLastBlockSize(INodeFileUnderConstruction pendingFile) {
    Block block = pendingFile.getLastBlock();
    if (block != null) {
      block.setNumBytes(pendingFile.getPreferredBlockSize());
    }
  }
1335611
private final void contendedLock() {
            boolean waiting = false;
            for (int s;;) {
                if (((s = lockState) & ~WAITER) == 0) {
                    if (U.compareAndSwapInt(this, LOCKSTATE, s, WRITER)) {
                        if (waiting)
                            waiter = null;
                        return;
                    }
                }
                else if ((s & WAITER) == 0) {
                    if (U.compareAndSwapInt(this, LOCKSTATE, s, s | WAITER)) {
                        waiting = true;
                        waiter = Thread.currentThread();
                    }
                }
                else if (waiting)
                    LockSupport.park(this);
            }
        }
998816
void metaSave(PrintWriter out) {
    synchronized (pendingReplications) {
      out.println("Metasave: Blocks being replicated: " +
                  pendingReplications.size());
      Iterator<Map.Entry<BlockInfo, PendingBlockInfo>> iter = 
          pendingReplications.entrySet().iterator();
      while (iter.hasNext()) {
        Map.Entry<BlockInfo, PendingBlockInfo> entry = iter.next();
        PendingBlockInfo pendingBlock = entry.getValue();
        BlockInfo block = entry.getKey();
        out.println(block + 
                    " StartTime: " + new Time(pendingBlock.timeStamp) +
                    " NumReplicaInProgress: " + 
                    pendingBlock.numReplicasInProgress);
      }
    }
  }
87761
public void lock() {
		if (cancel != null)
			return;
		if (error != null)
			return;
		Thread t;
		BlockedThreadHandler blockedHandler;
		do {
			synchronized (this) {
				if (!locked) {
					locked = true;
					return;
				}
				t = Thread.currentThread();
				blockedHandler = Threading.getBlockedThreadHandler(t);
				if (blockedHandler != null) break;
				try { this.wait(0); }
				catch (InterruptedException e) { continue; }
			}
		} while (true);
		blockedHandler.blocked(this, 0);
	}
1334522
public void sync() throws IOException {
  	long start = System.currentTimeMillis();
    try {
      long toWaitFor;
      synchronized (this) {
        eventStartSync();
        /* Record current blockOffset. This might be changed inside
         * flushBuffer() where a partial checksum chunk might be flushed.
         * After the flush, reset the bytesCurBlock back to its previous value,
         * any partial checksum chunk will be sent now and in next packet.
         */
        long saveOffset = bytesCurBlock;
        DFSOutputStreamPacket oldCurrentPacket = currentPacket;

        // flush checksum buffer as an incomplete chunk
        flushBuffer(false, shouldKeepPartialChunkData());
        // bytesCurBlock potentially incremented if there was buffered data
        
        eventSyncStartWaitAck();
        
        if (DFSClient.LOG.isDebugEnabled()) {
          DFSClient.LOG.debug("DFSClient flush() : bytesCurBlock " + bytesCurBlock +
                    " lastFlushOffset " + lastFlushOffset);
        }

        // Flush only if we haven't already flushed till this offset.
        if (lastFlushOffset != bytesCurBlock) {
          assert bytesCurBlock > lastFlushOffset;
          // record the valid offset of this flush
          lastFlushOffset = bytesCurBlock;
          enqueueCurrentPacket();
        } else {
          // just discard the current packet since it is already been sent.
          if (oldCurrentPacket == null && currentPacket != null) {
            // If we didn't previously have a packet queued, and now we do,
            // but we don't plan on sending it, then we should not
            // skip a sequence number for it!
            currentSeqno--;
          }
          currentPacket = null;
        }

        if (shouldKeepPartialChunkData()) {
          // Restore state of stream. Record the last flush offset
          // of the last full chunk that was flushed.
          //
          bytesCurBlock = saveOffset;
        }
        toWaitFor = lastQueuedSeqno;
      }

      waitForAckedSeqno(toWaitFor);
      
      eventSyncPktAcked();

      // If any new blocks were allocated since the last flush,
      // then persist block locations on namenode.
      //
      boolean willPersist;
      synchronized (this) {
        willPersist = persistBlocks;
        persistBlocks = false;
      }
      if (willPersist) {
        dfsClient.namenode.fsync(src, dfsClient.clientName);
      }
      long timeval = System.currentTimeMillis() - start;
      dfsClient.metrics.incSyncTime(timeval);

      eventEndSync();

    } catch (IOException e) {
        lastException = new IOException("IOException flush:", e);
        closed = true;
        closeThreads();
        throw e;
    }
  }
327347
function BlockData(index, buf, cb, count) {
  this.index = index;
  this.buf = buf;
  this.cb = cb;
  this.count = count | 0;
}
777254
protected void blockFinished (SceneBlock block)
    {
        --_pendingBlocks;

        // once all the visible pending blocks have completed their
        // resolution, recompute our visible object set and show ourselves
        if (_visiBlocks.remove(block) && _visiBlocks.size() == 0) {
            allBlocksFinished();
        }
    }
497562
def _pending_of(self, workload):
        """"""
        pending = sum(list(scope.values()).count(False) for scope in workload.values())
        return pending
714383
private void waitIndexingNotSuspended() {
        boolean suspendedNotified = false;
        synchronized (suspendLock) {
            while (suspend > 0) {
                if (!suspendedNotified) {
                    suspendedNotified = true;
                }
                try {
                    suspendLock.wait();
                } catch (InterruptedException ex) {
                }
            }
        }
    }
165866
public void lock() {
		if (cancel != null)
			return;
		if (error != null)
			return;
		Thread t = Thread.currentThread();
		if (t == lockingThread) {
			lockedTimes++;
			return;
		}
		BlockedThreadHandler blockedHandler = null;
		do {
			synchronized (this) {
				if (lockingThread == null) {
					lockingThread = t;
					lockedTimes = 1;
					return;
				}
				if (blockedHandler == null)
					blockedHandler = Threading.getBlockedThreadHandler(t);
				if (blockedHandler == null) {
					try { this.wait(0); }
					catch (InterruptedException e) { /* ignore */ }
					continue;
				}
			}
			blockedHandler.blocked(this, 0);
		} while (true);
	}
1212803
@Override
    public void flushQueueBlocking() {
        flushQueue();

        val context =((CudaContext) AtomicAllocator.getInstance().getDeviceContext().getContext());

        context.syncSpecialStream();
        context.syncOldStream();
    }
980953
public void setThisData(byte[] data)
    {
        try
        {
            pen.setData(data);
        }
        catch ( Exception e )
        {
            ThreadUtils.checkInterrupted(e);
            Throwables.propagate(e);
        }
    }
1255424
private void safeCustomDelay(PollingStrategyContext pollingStrategyContext) {
        try {
            pollingStrategy.getDelayStrategy().delayBeforeNextRetry(pollingStrategyContext);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new RuntimeException(e);
        }
    }
366292
def submit_and_block(meth, *args)
      @mutex.synchronize do
        raise Exceptions::NotConnected if unhealthy?
      end

      cnt = Continuation.new(meth, *args)
      @reg.synchronize do |r|
        if meth == :state
          r.state_check << cnt
        else
          r.pending << cnt
        end
      end
      wake_event_loop!
      cnt.value
    end
835663
private void notifyReadySendListeners(BlockingQueue<Pair<SendListener, Conversation>> readySendCallbacks) {
      if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) SibTr.entry(this, tc, "notifyReadySendListeners", readySendCallbacks);
      try {
         for (Pair<SendListener, Conversation> callback : readySendCallbacks) {
            callback.left.dataSent(callback.right);
         }
      } catch (Throwable t) {
         FFDCFilter.processException
                 (t, "com.ibm.ws.sib.jfapchannel.impl.ConnectionWriteCompletedCallback", JFapChannelConstants.CONNWRITECOMPCALLBACK_COMPLETE_01, connection.getDiagnostics(true));
         if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled())
            SibTr.debug(tc, "exception invoking send listener data sent");
         if (TraceComponent.isAnyTracingEnabled() && tc.isEventEnabled()) SibTr.exception(tc, t);
         connection.invalidate(true, t, "send listener threw exception");  // D224570
      }
      if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) SibTr.exit(this, tc, "notifyReadySendListeners");
   }