runs synchronization of a particular {@link Bucket}
321512
public void run() {
        AbstractKVStorable[] linkData = bucket.getBackend(); // get all LinkData

        long startTime = System.nanoTime(); // remember the, time when synchronizing is started (for logging)

        log.debug("Start to synchronize {} objects.", linkData.length);
        try {
            String filename = buffer.bucketContainer.getHashFunction().getFilename(bucket.getBucketId());
            String directoryName = buffer.getPathToDbFiles();
            Synchronizer<Data> synchronizer = synchronizerFactory
                    .createSynchronizer(directoryName + "/" + filename, gp);
            synchronizer.upsert(linkData); // start synchronizing

            actualProcessingBuckets.remove(bucket);
            freeMemory(bucket);
            synchronizer.close();
            log.debug("Synchronized {} objects in {} ms.", linkData.length, ((System.nanoTime() - startTime) / 1e6));
            /* update messages */
            buffer.sumUpInserted(synchronizer.getNumberOfInsertedEntries());
            buffer.sumUpUpdated(synchronizer.getNumberOfUpdatedEntries());
        } catch (Exception ex) {
            log.error("An error occurred during synchronizing. Synchronizing thread stopped! Some data was lost", ex);
            freeMemory(bucket);
            actualProcessingBuckets.remove(bucket);

        }
    }
3123
@Override
	protected void flush() {
		// The Kafka 0.8 producer doesn't support flushing, we wait here
		// until all pending records are confirmed
		synchronized (pendingRecordsLock) {
			while (pendingRecords > 0) {
				try {
					pendingRecordsLock.wait();
				} catch (InterruptedException e) {
					// this can be interrupted when the Task has been cancelled.
					// by throwing an exception, we ensure that this checkpoint doesn't get confirmed
					throw new RuntimeException("Flushing got interrupted while checkpointing", e);
				}
			}
		}
	}
24871
public Bucket reload() {
    // [START reload]
    Bucket latestBucket = bucket.reload(BucketSourceOption.metagenerationMatch());
    if (latestBucket == null) {
      // the bucket was not found
    }
    // [END reload]
    return latestBucket;
  }
307520
private Object syncGetValueFromBucket(FastSyncHashBucket hb, int key, boolean remove) {
        FastSyncHashEntry e = null;
        FastSyncHashEntry last = null;

        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) {
            Tr.entry(tc, "syncGetValueFromBucket: key, remove " + key + " " + remove);
        }
        synchronized (hb) {
            e = hb.root;
            while (e != null) {
                if (e.key == key) {
                    if (remove) {
                        if (last == null) {
                            // this is root
                            hb.root = e.next;
                        } else {
                            last.next = e.next;
                        }
                    }
                    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) {
                        Tr.exit(tc, "syncGetValueFromBucket: found value in bucket");
                    }
                    return e.value;
                }
                last = e;
                e = e.next;
            }
        }
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) {
            Tr.exit(tc, "syncGetValueFromBucket-returned null");
        }
        return null;
    }
1068144
void lockBucketsRead(long i1, long i2) {
		int bucket1LockIdx = getBucketLock(i1);
		int bucket2LockIdx = getBucketLock(i2);
		// always lock segments in same order to avoid deadlocks
		if (bucket1LockIdx < bucket2LockIdx) {
			lockAry[bucket1LockIdx].readLock();
			lockAry[bucket2LockIdx].readLock();
		} else if (bucket1LockIdx > bucket2LockIdx) {
			lockAry[bucket2LockIdx].readLock();
			lockAry[bucket1LockIdx].readLock();
		}
		// if we get here both indexes are on same segment so only lock once!!!
		else {
			lockAry[bucket1LockIdx].readLock();
		}
	}
709270
private void syncFromStorage()
  {
    giant.lock();

    try {
      if (active) {
        final Map<String, Task> newTasks = toTaskIDMap(taskStorage.getActiveTasks());
        final int tasksSynced = newTasks.size();
        final Map<String, Task> oldTasks = toTaskIDMap(tasks);

        // Calculate differences on IDs instead of Task Objects.
        Set<String> commonIds = Sets.newHashSet(Sets.intersection(newTasks.keySet(), oldTasks.keySet()));
        for (String taskID : commonIds) {
          newTasks.remove(taskID);
          oldTasks.remove(taskID);
        }
        Collection<Task> addedTasks = newTasks.values();
        Collection<Task> removedTasks = oldTasks.values();

        // Clean up removed Tasks
        for (Task task : removedTasks) {
          removeTaskInternal(task);
        }

        // Add newly Added tasks to the queue
        for (Task task : addedTasks) {
          addTaskInternal(task);
        }

        log.info(
            "Synced %d tasks from storage (%d tasks added, %d tasks removed).",
            tasksSynced,
            addedTasks.size(),
            removedTasks.size()
        );
        managementMayBeNecessary.signalAll();
      } else {
        log.info("Not active. Skipping storage sync.");
      }
    }
    catch (Exception e) {
      log.warn(e, "Failed to sync tasks from storage!");
      throw new RuntimeException(e);
    }
    finally {
      giant.unlock();
    }
  }
284600
public void run() throws Exception {
    try {

      CountDownLatch countDownLatch = new CountDownLatch(this.tasks);
      for (int i = 0; i < this.tasks; i++) {
        addTask(i, countDownLatch);
      }
      // Wait for the tasks to finish
      countDownLatch.await();
    } finally {
      try {
        // Calling close() will stop metric reporting
        this.context.close();
      } finally {
        this.executor.shutdownNow();
      }
    }
  }
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);
	}
780626
protected void rolling() {
        synchronized (this) {
            if (unflushed != null) {
                buckets.put(currBucketTime, unflushed);
                unflushed = null;
            }

            currBucketTime = getCurrBucketTime();

            return;
        }
    }
358774
public Bucket update() {
    // [START update]
    Bucket updatedBucket = bucket.toBuilder().setVersioningEnabled(true).build().update();
    // [END update]
    return updatedBucket;
  }
523583
public void finishSynch() throws IOException {
		if (finishing != null) {
			finishing.blockException(0);
			return;
		}
		finishing = new SynchronizationPoint<>();
		try {
			AsyncWork<Integer, IOException> lastWrite = writeOps.getLastPendingOperation();
			if (lastWrite != null) {
				lastWrite.blockException(0);
			}
			deflater.finish();
			if (!deflater.finished()) {
				byte[] writeBuf = new byte[8192];
				do {
					int nb = deflater.deflate(writeBuf, 0, writeBuf.length);
					if (nb <= 0) break;
					output.writeSync(ByteBuffer.wrap(writeBuf, 0, nb));
				} while (!deflater.finished());
			}
			finishing.unblock();
		} catch (IOException e) {
			finishing.error(e);
			throw e;
		}
	}
365964
public void sync() {
		if (!validState) {
			throw new InvalidStateException();
		}
		if (useMmap) {
			syncAllMmaps();
		}
		if (fileChannel != null) {
			try {
				fileChannel.force(false);
			} catch (Exception ign) {
			}
		}
		if (callback != null) {
			callback.synched();
		}
	}
360103
@Override
  public void run() {
    synchronized (this) {
      started = true;
    }
    finished = false;

    final byte[] buf = new byte[bufferSize];

    try {
      int length;
      while (true) {
        waitForInput(is);

        if (finish || Thread.interrupted()) {
          break;
        }

        length = is.read(buf);
        if (length <= 0 || Thread.interrupted()) {
          break;
        }
        os.write(buf, 0, length);
        if (autoflush) {
          os.flush();
        }
        if (finish) { //NOSONAR
          break;
        }
      }
      // On completion, drain any available data (which might be the first data available for quick executions)
      if (finish) {
        while((length = is.available()) > 0) {
          if (Thread.interrupted()) {
            break;
          }
          length = is.read(buf, 0, Math.min(length, buf.length));
          if (length <= 0) {
            break;
          }
          os.write(buf, 0, length);
        }
      }
      os.flush();
    } catch (InterruptedException ie) {
      // likely PumpStreamHandler trying to stop us
    } catch (Exception e) {
      synchronized (this) {
        exception = e;
      }
    } finally {
      if (closeWhenExhausted) {
        Closeables.close(os);
      }
      finished = true;
      finish = false;
      synchronized (this) {
        notifyAll();
      }
    }
  }
936379
public SyncLazyFuture<X> withRunningThread() {
        if (state != null) return this;
        Thread thread = new Thread(SyncLazyFuture.this.getClass().getName() + ".withRunningThread") {
            @Override
            public void run() {
                try {
                    SyncLazyFuture.this.internalCalc();
                } catch (ExecutionException | InterruptedException e) {
                    e.printStackTrace();
                }
            }
        };
        thread.start();
        return this;
    }
1006859
public void sync(final String path, VoidCallback cb, Object ctx) {
        verbotenThreadCheck();
        final String clientPath = path;
        PathUtils.validatePath(clientPath);

        final String serverPath = prependChroot(clientPath);

        RequestHeader h = new RequestHeader();
        h.setType(ZooDefs.OpCode.sync);
        SyncRequest request = new SyncRequest();
        SyncResponse response = new SyncResponse();
        request.setPath(serverPath);
        cnxn.queuePacket(h, new ReplyHeader(), request, response, cb,
                clientPath, serverPath, ctx, null);
    }
1128716
@Override
    public void run() {
        while(_enabled) {
            if(_newCycle.compareAndSet(true, false)) {
                // One and only one compactor is at work.
                _lock.lock();
                
                try {
                    reset();
                    _state = State.INIT;
                    _log.info("cycle init");
                    
                    // Flush segment index buffers
                    flushSegmentIndexBuffers();
                    
                    // Free compacted segments
                    freeCompactedSegments();
                    
                    // Inspect the array
                    if(!inspect()) continue;
                    
                    // Compact the array
                    if(!compact()) continue;
                } catch(Exception e) {
                    _log.error("compaction failure", e);
                } finally {
                    reset();
                    _state = State.DONE;
                    _log.info("cycle done");
                    _lock.unlock();
                }
            } else {
                try {
                    Thread.sleep(100);
                } catch (InterruptedException e) {
                    _log.warn(e.getMessage());
                }
            }
        }
    }
1081069
private synchronized void initBuckets() {
    final SparseIntArray bucketSizes = mPoolParams.bucketSizes;

    // create the buckets
    if (bucketSizes != null) {
      fillBuckets(bucketSizes);
      mAllowNewBuckets = false;
    } else {
      mAllowNewBuckets = true;
    }
  }