eventually consistent processing
@param logCtx
178089
public boolean process(LogProcessContext logCtx){
		
		Boolean currentTrxStatus = logCtx.getFinalMasterTransStatus();
		logCtx.getMasterTransactionStatusVotter().setTransactionStatus(currentTrxStatus);
		
		LogCollection logCollection = logCtx.getLogCollection();
		List<Content> orderedContents = logCollection.getOrderedContents();
		
		//依次调用LOG对应的processor里的preProcess
		//目前preProcess中其中一个任务就是确定主控事务状态，另外一个就是执行SAGA的 正向调用/TRY 等方法
		//若主控事务状态未知，则部分LOG对应的preProcess操作里将会投票表决主控事务状态，若有表示不能提交的话，则事务状态为回滚
		//根据preProcess的数据更新主控事务状态（主控事务状态确定，才能往下执行，）
		for(int i = 0;i < orderedContents.size() ; i++){
			Content content = orderedContents.get(i);
			//check log order
			Assert.isTrue(content.getcId() != null && content.getcId().equals(i + 1),"content list did not sort or contentId is null");
			
			Class<? extends LogProcessor> proccessorClass = ContentType.getById(content.getLogType()).getProccessorClass();
			if(proccessorClass == null){
				if(LOG.isDebugEnabled()){
					LOG.debug("processor not set,continue" + content);
				}
				continue;
			}
			
			LogProcessor processor = proccessorMap.get(proccessorClass);
			if(!processor.preLogProcess(logCtx, content)){
				LOG.warn( "log pre-Processor return false,end proccesing and retry later." + content);
				return false;
			}
		}
		
		// 事务状态未知，且本事务为主控事务，则更新主控事务状态
		if (currentTrxStatus == null
				&& MetaDataFilter.getMetaData(EasytransConstant.CallHeadKeys.PARENT_TRX_ID_KEY) == null) {
			boolean allowCommit = logCtx.getMasterTransactionStatusVotter().getCurrentVoteStatusCommited();
			int updateCount = transChecker.updateMasterTransactionStatus(logCtx.getTransactionId(),
					allowCommit ? TransactionStatus.COMMITTED : TransactionStatus.ROLLBACKED);
			
			//concurrent modify check 
			if(updateCount == 0) {
				throw new RuntimeException("can not find the trx,or the status of Transaction is not UNKOWN!");
			}
			
			logCtx.setFinalMasterTransStatus(allowCommit);
		}
		
		//依次调用LOG对应的Processor的logProcess
		//统计并发布SemiLog事件
		//发布ProcessEnd事件
		for(int i = 0;i < orderedContents.size() ; i++){
			
			Content content = orderedContents.get(i);
			//check log order
			Assert.isTrue(content.getcId() != null && content.getcId().equals(i + 1),"content list did not sort or contentId is null");
			
			if(content instanceof DemiLeftContent){
				logCtx.getDemiLogManager().registerLeftDemiLog((DemiLeftContent)content);
			}else if(content instanceof DemiRightContent){
				logCtx.getDemiLogManager().registerRightDemiLog((DemiRightContent)content);
			}
			
			Class<? extends LogProcessor> proccessorClass = ContentType.getById(content.getLogType()).getProccessorClass();
			if(proccessorClass == null){
				if(LOG.isDebugEnabled()){
					LOG.debug("processor not set,continue" + content);
				}
				continue;
			}
			
			LogProcessor processor = proccessorMap.get(proccessorClass);
			if(!processor.logProcess(logCtx, content)){
				LOG.warn( "log processor return false,end proccesing and retry later." + content);
				return false;
			}
		}

		if(!logCtx.getDemiLogManager().pubulishDemiLogEvent()){
			LOG.warn("DemiLogEvent Process return false,end proccesing and retry later." + logCollection);
		}
		
		if(!logCtx.getProcessEndManager().publish()){
			LOG.warn("End process return false,end proccesing and retry later." + logCollection);
		}
		
		//end and flush log
		if(leastLogModel){
			logCtx.getLogCache().clearCacheLogs();
		}
		logCtx.getLogCache().flush(true);
		return true;
	}
526661
private void afterBatch(BatchBackend backend) {
		IndexedTypeSet targetedTypes = searchFactoryImplementor.getIndexedTypesPolymorphic( rootIndexedTypes );
		if ( this.optimizeAtEnd ) {
			backend.optimize( targetedTypes );
		}
		backend.flush( targetedTypes );
	}
128531
@Override @CallerSideMethod public void onNext(IN in) {
        if ( in  == null )
            throw null;
        _onNext(in);
    }
533188
public synchronized void flush(boolean bSendFakeTrx)
    {
        try {
            if (bSendFakeTrx)
                this.getWriter().writeObject(FAKE_TRX);
            this.getWriter().flush();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
58639
def runctx(cmd, globals, locals, filename=None, threads=True, verbose=False):
    """"""
    _run(threads, verbose, 'runctx', filename, cmd, globals, locals)
772118
@Override
    public void upgradeFrom(ChannelHandlerContext ctx) {
        final ChannelPipeline p = ctx.pipeline();
        p.remove(this);
    }
579198
public void postProcess(DirContext ctx) throws NamingException {
        for (DirContextProcessor processor : dirContextProcessors) {
            processor.postProcess(ctx);
        }
    }
1026966
private void initialDecodeTasks(final ChannelHandlerContext ctx) {
        currentRequest = sentRequestQueue.poll();
        currentDecodingState = DecodingState.STARTED;

        if (currentRequest != null) {
            Long st = sentRequestTimings.poll();
            if (st != null) {
                currentOpTime = System.nanoTime() - st;
            } else {
                currentOpTime = -1;
            }
        }

        if (env().operationTracingEnabled()) {
            Span dispatchSpan = dispatchSpans.poll();
            if (dispatchSpan != null) {
                currentDispatchSpan = dispatchSpan;
            }
        }

        if (traceEnabled) {
            LOGGER.trace("{}Started decoding of {}", logIdent(ctx, endpoint), currentRequest);
        }
    }
1243539
@Override
	public void process(Packet message, MessageContext ctx)
	{
		EntityPlayerSP player = (EntityPlayerSP) Utils.getClientPlayer();
		if (message.type == ContainerType.TYPE_TILEENTITY)
		{
			IDirectInventoryProvider inventoryProvider = TileEntityUtils.getTileEntity(	IDirectInventoryProvider.class,
																						Utils.getClientWorld(),
																						message.pos);
			if (inventoryProvider != null)
				MalisisInventory.open(player, inventoryProvider, message.windowId);
		}

		else if (message.type == ContainerType.TYPE_ITEM)
		{
			//TODO: send and use slot number instead of limited to equipped
			ItemStack itemStack = player.getHeldItemMainhand();
			if (itemStack == null || !(itemStack.getItem() instanceof IDeferredInventoryProvider<?>))
				return;

			@SuppressWarnings("unchecked")
			IDeferredInventoryProvider<ItemStack> inventoryProvider = (IDeferredInventoryProvider<ItemStack>) itemStack.getItem();
			MalisisInventory.open(player, inventoryProvider, itemStack, message.windowId);
		}
	}
473568
public synchronized void logTrx(Object objLogData)
    {
        BaseMessage message = this.getMessageObject();
        message.putNative(MESSAGE_PARAM, objLogData);
        this.getMessageManager().sendMessage(message);
    }
1190769
@Override
	public void process(Packet message, MessageContext ctx)
	{
		EntityPlayerSP player = (EntityPlayerSP) Utils.getClientPlayer();
		Container c = player.openContainer;
		if (message.windowId != c.windowId || !(c instanceof MalisisInventoryContainer))
			return;

		MalisisInventoryContainer container = (MalisisInventoryContainer) c;
		if (message.inventoryId == PICKEDITEM)
		{
			container.setPickedItemStack(message.slots.get(-1));
			return;
		}

		MalisisInventory inventory = container.getInventory(message.inventoryId);
		if (inventory == null)
			return;

		for (Entry<Integer, ItemStack> entry : message.slots.entrySet())
		{
			Integer slotNumber = entry.getKey();
			ItemStack itemStack = entry.getValue();

			inventory.setItemStack(slotNumber, itemStack);
		}
	}
535195
private void processRx(byte[] buff, int len) {
        int i = 0;
        // start HDLC alg
        while (i < len) {
            while (rxState.bits <= 24 && i < len) {
                int b = buff[i++] & 0xff;
                hdlc.fasthdlc_rx_load_nocheck(rxState, b);
                if (rxState.state == 0) {
                    // octet counting mode
                    nCount = (nCount + 1) % 16;
                    if (nCount == 0) {
                        countError("on receive");
                    }
                }
            }

            int res = hdlc.fasthdlc_rx_run(rxState);

            switch (res) {
                case FastHDLC.RETURN_COMPLETE_FLAG:
                    // frame received and we count it
                    countFrame();

                    // checking length and CRC of the received frame
                    if (rxFrame.len == 0) {
                    } else if (rxFrame.len < 5) {
                        // frame must be at least 5 bytes in length
                        countError("hdlc error, frame LI<5");
                    } else if (rxCRC == 0xF0B8) {
                        // good frame received
                        processFrame();
                    } else {
                        countError("hdlc complete, wrong terms.");
                    }
                    rxFrame.len = 0;
                    rxCRC = 0xffff;
                    break;
                case FastHDLC.RETURN_DISCARD_FLAG:
                    // looking for next flag
                    rxCRC = 0xffff;
                    rxFrame.len = 0;
                    // eCount = 0;
                    countFrame();
                    // "on receive, hdlc discard"
                    countError("hdlc discard.");
                    break;
                case FastHDLC.RETURN_EMPTY_FLAG:
                    rxFrame.len = 0;
                    break;
                default:
                    if (rxFrame.len > 279) {
                        rxState.state = 0;
                        rxFrame.len = 0;
                        rxCRC = 0xffff;
                        eCount = 0;
                        countFrame();
                        countError("Overlong MTP frame, entering octet mode on link '" + name + "'");
                    } else {
                        rxFrame.frame[rxFrame.len++] = (byte) res;
                        rxCRC = PPP_FCS(rxCRC, res & 0xff);
                    }
            }
        }
    }
725870
private void processNow() {
        List<T> toProcess;
        boolean scheduleNextBatchImmediately = false;
        synchronized (mutex) {
            int count = inbox.size();
            Log.v(Log.TAG_BATCHER, "%s: processNow() called, inbox size: %d", this, count);
            if (count == 0)
                return;
            else if (count <= capacity) {
                toProcess = new ArrayList<T>(inbox);
                inbox.clear();
            } else {
                toProcess = new ArrayList<T>(inbox.subList(0, capacity));
                for (int i = 0; i < capacity; i++)
                    inbox.remove(0);
                scheduleNextBatchImmediately = true;
            }
            mutex.notifyAll();
        }

        synchronized (processMutex) {
            if (toProcess != null && toProcess.size() > 0) {
                Log.v(Log.TAG_BATCHER, "%s: invoking processor %s with %d items",
                        this, processor, toProcess.size());
                processor.process(toProcess);
            } else
                Log.v(Log.TAG_BATCHER, "%s: nothing to process", this);

            synchronized (mutex) {
                lastProcessedTime = System.currentTimeMillis();
                scheduled = false;
                scheduleBatchProcess(scheduleNextBatchImmediately);
                Log.v(Log.TAG_BATCHER, "%s: invoking processor done",
                        this, processor, toProcess.size());
            }
            processMutex.notifyAll();
        }
    }
1125429
private void updateState(FSEditLogOp op, boolean checkTxnId) throws IOException {
    InjectionHandler.processEvent(InjectionEvent.SERVERLOGREADER_UPDATE, op);
    
    if (checkTxnId) {
      mostRecentlyReadTransactionTxId = ServerLogReaderUtil.checkTransactionId(
        mostRecentlyReadTransactionTxId, op);
    }
    
    updateStreamPosition();

    // read a valid operation
    core.getMetrics().readOperations.inc();
    mostRecentlyReadTransactionTime = now();

    // current log segment ends normally
    if (op.opCode == FSEditLogOpCodes.OP_END_LOG_SEGMENT) {
      LOG.info("Segment - ending log segment start txid: " + currentSegmentTxId
          + ", end txid: " + op.getTransactionId());
      // move forward with next segment
      currentSegmentTxId = op.getTransactionId() + 1;
      // set the stream to null so the next getNotification()
      // will recreate it
      currentEditLogInputStream = null;
      // indicate that a new stream will be opened
      currentEditLogInputStreamPosition = -1;
    } else if (op.opCode == FSEditLogOpCodes.OP_START_LOG_SEGMENT) {
      LOG.info("Segment - starting log segment start txid: "
          + currentSegmentTxId);
    }
  }
655067
void processPersistenceUnit(JPAPXml pxml, JPAApplInfo applInfo) {
        final boolean isTraceOn = TraceComponent.isAnyTracingEnabled();
        if (isTraceOn && tc.isEntryEnabled())
            Tr.entry(tc, "processPersistenceUnit" + pxml);

        synchronized (pxmlsInfo) {
            URL pxmlRootURL = pxml.getRootURL();
            String pxmlInfoKey = scope == JPAPuScope.EAR_Scope ? pxml.getArchiveName() : pxmlRootURL.toString();
            JPAPxmlInfo pxmlInfo = pxmlsInfo.get(pxmlInfoKey);

            // if root URL is defined for this persistence.xml, it has been processed before,
            // i.e. deployed web module processing will find persistence.xml in the ejb module
            // since the ejb module is in the web module classpath.
            if (pxmlInfo == null) {
                pxmlInfo = new JPAPxmlInfo(this, pxmlRootURL); // F743-23167
                pxmlsInfo.put(pxmlInfoKey, pxmlInfo);

                JPAIntrospection.beginPXmlInfoVisit(pxmlInfo);
                try {
                    // read in all the persistence unit define in this persistence.xml
                    pxmlInfo.extractPersistenceUnits(pxml); //PK62950
                } finally {
                    JPAIntrospection.endPXmlInfoVisit();
                }
            }

        }

        if (isTraceOn && tc.isEntryEnabled())
            Tr.exit(tc, "processPersistenceUnit", getPUList());
    }
1287769
synchronized void endCurrentLogSegment(boolean writeEndTxn)
      throws IOException {
    LOG.info("Ending log segment " + curSegmentTxId);
    if (state != State.IN_SEGMENT) {
      throw new IllegalStateException("Bad state: " + state);
    }
    waitForSyncToFinish();
    if (writeEndTxn) {
      logEdit(LogSegmentOp.getInstance(FSEditLogOpCodes.OP_END_LOG_SEGMENT));
    }
    logSyncAll();
    printStatistics(true);
    final long lastTxId = getLastWrittenTxId();

    try {
      journalSet.finalizeLogSegment(curSegmentTxId, lastTxId);
      editLogStream = null;
    } catch (IOException e) {
      // All journals have failed, it will be handled in logSync.
      FSNamesystem.LOG.info("Cannot finalize log segment: " + e.toString());
    }
    state = State.BETWEEN_LOG_SEGMENTS;
  }
543042
def flush(self):
        """"""
        if self.__logFileStream is not None:
            try:
                self.__logFileStream.flush()
            except:
                pass
            try:
                os.fsync(self.__logFileStream.fileno())
            except:
                pass
        if self.__stdout is not None:
            try:
                self.__stdout.flush()
            except:
                pass
            try:
                os.fsync(self.__stdout.fileno())
            except:
                pass