<commit_before>/**
 * 
 */
package actorx.util;

import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * @author Xiong
 */
public class MpscQueue<E> {
	private ConcurrentLinkedQueue<E> que = new ConcurrentLinkedQueue<E>();
	private AtomicBoolean blocked = new AtomicBoolean(false);

	public void add(E e){
		que.add(e);
	}

	/**
	 * @return E or null
	 */
	public E poll(){
		return que.poll();
	}
	
	/**
	 * @param e
	 */
	public void put(E e){
		add(e);
		// 如果发现单读线程在阻塞，唤醒它
		if (isBlocked()){
			synchronized (que){
				if (isBlocked()){
					que.notify();
				}
			}
		}
	}
	
	/**
	 * 警告：只能在单读线程调用
	 * @return E or null
	 */
	public E take(){
		assert(!isBlocked());
		E e = poll();
		if (e != null){
			return e;
		}

		synchronized (que){
			block();
			e = poll();
			while (e == null){
				try{
					que.wait();
				}catch (InterruptedException ex){}
				
				e = poll();
			}
		}

		unblock();
		return e;
	}
	
	/**
	 * 警告：只能在单读线程调用
	 * @param timeout
	 * @return E or null
	 */
	public E poll(long timeout){
		assert(!isBlocked());
		if (timeout <= 0){
			return poll();
		}
		
		if (timeout == Long.MAX_VALUE){
			return take();
		}
		
		E e = poll();
		if (e != null){
			return e;
		}

		synchronized (que){
			block();
			e = poll();
			long currTimeout = timeout;
			while (e == null){
				long bt = System.currentTimeMillis();
				try{
					que.wait(currTimeout);
				}catch (InterruptedException ex){}
				long eclipse = System.currentTimeMillis() - bt;
				
				e = poll();
				if (eclipse >= currTimeout){
					break;
				}
				
				currTimeout -= eclipse;
			}
		}

		unblock();
		return e;
	}
	
	private boolean isBlocked(){
		return blocked.get();
	}
	
	private void block(){
		blocked.set(true);
	}
	
	private void unblock(){
		blocked.set(false);
	}
}
<commit_msg>improve MspcQueue change blocked to volatile not atomicboolean<commit_after>/**
 * 
 */
package actorx.util;

import java.util.concurrent.ConcurrentLinkedQueue;

/**
 * @author Xiong
 */
public class MpscQueue<E> {
	private ConcurrentLinkedQueue<E> que = new ConcurrentLinkedQueue<E>();
	private volatile boolean blocked = false;

	public void add(E e){
		que.add(e);
	}

	/**
	 * @return E or null
	 */
	public E poll(){
		return que.poll();
	}
	
	/**
	 * @param e
	 */
	public void put(E e){
		add(e);
		// 如果发现单读线程在阻塞，唤醒它
		if (isBlocked()){
			synchronized (que){
				if (isBlocked()){
					que.notify();
				}
			}
		}
	}
	
	/**
	 * 警告：只能在单读线程调用
	 * @return E or null
	 */
	public E take(){
		assert(!isBlocked());
		E e = poll();
		if (e != null){
			return e;
		}

		synchronized (que){
			block();
			e = poll();
			while (e == null){
				try{
					que.wait();
				}catch (InterruptedException ex){}
				
				e = poll();
			}
		}

		unblock();
		return e;
	}
	
	/**
	 * 警告：只能在单读线程调用
	 * @param timeout
	 * @return E or null
	 */
	public E poll(long timeout){
		assert(!isBlocked());
		if (timeout <= 0){
			return poll();
		}
		
		if (timeout == Long.MAX_VALUE){
			return take();
		}
		
		E e = poll();
		if (e != null){
			return e;
		}

		synchronized (que){
			block();
			e = poll();
			long currTimeout = timeout;
			while (e == null){
				long bt = System.currentTimeMillis();
				try{
					que.wait(currTimeout);
				}catch (InterruptedException ex){}
				long eclipse = System.currentTimeMillis() - bt;
				
				e = poll();
				if (eclipse >= currTimeout){
					break;
				}
				
				currTimeout -= eclipse;
			}
		}

		unblock();
		return e;
	}
	
	private boolean isBlocked(){
		return blocked;
	}
	
	private void block(){
		blocked = true;
	}
	
	private void unblock(){
		blocked = false;
	}
}

{'fraction_non_alphanumeric': 0.16087551299589603, 'fraction_numerical': 0.0005471956224350205, 'mean_word_length': 2.4654028436018955, 'pattern_counts': {'":': 0, '<': 11, '<?xml version=': 0, '>': 11, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5692899', 'n_tokens_mistral': 1678, 'n_tokens_neox': 1385, 'n_words': 333}