Deprecate the current batch and create a new one. The old one will still
be active until all writers have finished writing their current blocks
to it.
549357
void roll() {
        if (m_logger.isTraceEnabled()) {
            m_logger.trace("Rolling batch.");
        }
        Map<RollingDecoder,HttpExportDecoder> decoderMap;
        synchronized(m_tableDecoders) {
            decoderMap = ImmutableMap.copyOf(m_tableDecoders);
        }
        m_logger.info("Rolling " + decoderMap.size() + " number of data sources.");

        for (Map.Entry<RollingDecoder,HttpExportDecoder> entry : decoderMap.entrySet()) {
            entry.getValue().m_exportPath = null;
        }
    }
1223147
def chain_update(self, block, receipts):
        
        with self._lock:
            for batch_id in self._pending.copy():
                if self._batch_committed(batch_id):
                    self._pending.remove(batch_id)
                    self._update_observers(batch_id,
                                           ClientBatchStatus.COMMITTED)
499082
public static void cleanObsolete() {
        HandlerThreadFactory.getWriteLogThreadHandler().post(new Runnable() {
            @Override
            public void run() {
                long now = System.currentTimeMillis();
                File[] f = BlockCanaryInternals.getLogFiles();
                if (f != null && f.length > 0) {
                    synchronized (SAVE_DELETE_LOCK) {
                        for (File aF : f) {
                            if (now - aF.lastModified() > OBSOLETE_DURATION) {
                                aF.delete();
                            }
                        }
                    }
                }
            }
        });
    }
801009
def unpersist(self, blocking=False):
        
        self.is_cached = False
        self._jdf.unpersist(blocking)
        return self
703252
public void uninit() {
        if (initialized.get()) {
            log.debug("Uninit");
            if (writer != null) {
                if (writerFuture != null) {
                    try {
                        writerFuture.get();
                    } catch (Exception e) {
                        log.warn("Exception waiting for write result on uninit", e);
                    }
                    if (writerFuture.cancel(false)) {
                        log.debug("Future completed");
                    }
                }
                writerFuture = null;
                // write all the queued items
                doWrites();
                // clear the queue
                queue.clear();
                queue = null;
                // close the writer
                writer.close();
                writer = null;
            }
            // clear path ref
            path = null;
        }
    }
857816
def copy(self):
        """"""
        return self.__class__([b.copy() for b in self.blocks],
                tuple(self.pos) if self.pos else None)
420658
synchronized boolean remove(BlockInfo blockInfo, 
                              int oldReplicas, 
                              int decommissionedReplicas,
                              int oldExpectedReplicas) {
    int priLevel = getPriority(blockInfo, oldReplicas, 
                               decommissionedReplicas,
                               oldExpectedReplicas);
    return remove(blockInfo, priLevel);
  }
506513
def _abort_batches(self):
        """"""
        error = Errors.IllegalStateError("Producer is closed forcefully.")
        for batch in self._incomplete.all():
            tp = batch.topic_partition
            # Close the batch before aborting
            with self._tp_locks[tp]:
                batch.records.close()
            batch.done(exception=error)
            self.deallocate(batch)
369906
def rebuild( self ):
        
        plugins.init()
        
        self.blockSignals(True)
        self.setUpdatesEnabled(False)
        
        # clear the old editor
        if ( self._editor ):
            self._editor.close()
            self._editor.setParent(None)
            self._editor.deleteLater()
            self._editor = None
        
        # create a new widget
        plugin_class = plugins.widgets.get(self._columnType)
        if ( plugin_class ):
            self._editor = plugin_class(self)
            self.layout().addWidget(self._editor)
        
        self.blockSignals(False)
        self.setUpdatesEnabled(True)
864161
void discardUnusedTail()
    {
        // we guard this with the OpOrdering instead of synchronised due to potential dead-lock with CLSM.advanceAllocatingFrom()
        // this actually isn't strictly necessary, as currently all calls to discardUnusedTail occur within a block
        // already protected by this OpOrdering, but to prevent future potential mistakes, we duplicate the protection here
        // so that the contract between discardUnusedTail() and sync() is more explicit.
        try (OpOrder.Group group = appendOrder.start())
        {
            while (true)
            {
                int prev = allocatePosition.get();
                // we set allocatePosition past buffer.capacity() to make sure we always set discardedTailFrom
                int next = buffer.capacity() + 1;
                if (prev == next)
                    return;
                if (allocatePosition.compareAndSet(prev, next))
                {
                    discardedTailFrom = prev;
                    return;
                }
            }
        }
    }
755184
protected synchronized WriterPoolMember makeNewWriterIfAppropriate() {
	    long now = System.currentTimeMillis();
	    lastWriterNeededTime = now; 
        if(currentActive < maxActive) {
            currentActive++;
            lastWriterRolloverTime = now; 
            return makeWriter(); 
        }
        return null; 
    }
760515
def update(
        self,
        filename=None,
        batch_id=None,
        prev_batch_id=None,
        producer=None,
        count=None,
    ):
        
        # TODO: refactor model enforce unique batch_id
        # TODO: refactor model to not allow NULLs
        if not filename:
            raise BatchHistoryError("Invalid filename. Got None")
        if not batch_id:
            raise BatchHistoryError("Invalid batch_id. Got None")
        if not prev_batch_id:
            raise BatchHistoryError("Invalid prev_batch_id. Got None")
        if not producer:
            raise BatchHistoryError("Invalid producer. Got None")
        if self.exists(batch_id=batch_id):
            raise IntegrityError("Duplicate batch_id")
        try:
            obj = self.model.objects.get(batch_id=batch_id)
        except self.model.DoesNotExist:
            obj = self.model(
                filename=filename,
                batch_id=batch_id,
                prev_batch_id=prev_batch_id,
                producer=producer,
                total=count,
            )
            obj.transaction_file.name = filename
            obj.save()
        return obj
74748
def preserve_batch_state():
    
    with LOCK:
        old = ROOT.gROOT.IsBatch()
        try:
            yield
        finally:
            ROOT.gROOT.SetBatch(old)
467218
private void clear(Object caller, int version) {
        assert mutation_in_progress(caller, version);
        _buf_limit = 0;
        for (int ii=0; ii<_blocks.size(); ii++) {
            _blocks.get(ii).clearBlock();
            // _blocks.get(ii)._idx = -1; this is done in clearBlock()
        }
        bbBlock first = _blocks.get(0);
        first._idx = 0;                        // cas: 26 dec 2008
        first._offset = 0;
        first._limit = 0;
        _next_block_position = 1;
        return;
    }
78463
def update_next_block(self):
        
        last = self.mem[-1]
        if last.inst not in ('ret', 'jp', 'jr') or last.condition_flag is not None:
            return

        if last.inst == 'ret':
            if self.next is not None:
                self.next.delete_from(self)
                self.delete_goes(self.next)
            return

        if last.opers[0] not in LABELS.keys():
            __DEBUG__("INFO: %s is not defined. No optimization is done." % last.opers[0], 2)
            LABELS[last.opers[0]] = LabelInfo(last.opers[0], 0, DummyBasicBlock(ALL_REGS, ALL_REGS))

        n_block = LABELS[last.opers[0]].basic_block
        if self.next is n_block:
            return

        if self.next.prev == self:
            # The next basic block is not this one since it ends with a jump
            self.next.delete_from(self)
            self.delete_goes(self.next)

        self.next = n_block
        self.next.add_comes_from(self)
        self.add_goes_to(self.next)
608078
def cancel(self, block=True):
        
        if self._consumer:
            self._consumer.cancel(block=block)
345369
def _dequeue_batch(self) -> Optional[Batch]:
        
        if self._enqueueing_thread is None:
            raise ValueError('StreamWrapper `{}` with buffer of size `{}` was used outside with-resource environment.'
                             .format(self._name, self._buffer_size))
        if not self._enqueueing_thread.is_alive() and self._queue.empty():
            self._start_thread()
        while True:
            try:
                batch = self._queue.get(timeout=2)
                self._queue.task_done()
                break
            except Empty:
                if not self._enqueueing_thread.is_alive():
                    try:
                        # the enqueueing thread may just finished properly so lets check the queue eagerly
                        batch = self._queue.get_nowait()
                        self._queue.task_done()
                        break
                    except Empty:
                        # so we failed to retrieve a batch and the enqueueing thread is dead
                        # there is no hope, something must went wrong
                        raise ChildProcessError('Enqueueing thread ended unexpectedly.')
        return batch