<commit_before>/*
 * Copyright (c) 2016-2018 Vegard IT GmbH, https://vegardit.com
 * SPDX-License-Identifier: Apache-2.0
 */
package hx.concurrent.collection;

import hx.concurrent.atomic.AtomicInt;
import hx.concurrent.lock.RLock;
import hx.concurrent.thread.Threads;

/**
 * Unbound thread-safe first-in-first-out message queue.
 *
 * @author <NAME>, Vegard IT GmbH
 */
class Queue<T> {

    #if cpp
        var _queue = new cpp.vm.Deque<T>();
    #elseif (hl && haxe_ver >= 4)
        var _queue = new hl.vm.Deque<T>();
    #elseif neko
        var _queue = new neko.vm.Deque<T>();
    #elseif java
        var _queue = new java.util.concurrent.ConcurrentLinkedDeque<T>();
    #elseif python
        var _queue:Dynamic;
    #else
        var _queue = new List<T>();
        var _queueLock = new RLock();
    #end

    public var length(get, never):Int;
    var _length = new AtomicInt(0);
    inline function get_length():Int return _length;

    public function new() {
        #if python
            python.Syntax.pythonCode('import collections');
            _queue = untyped collections.deque();
        #end
    }


    #if threads
    /**
     * Pop a message from the queue head.
     *
     * By default (with timeoutMS=0) this function is non-blocking, meaning if no message is available in the queue
     * `null` is returned immediately.
     *
     * If <code>timeoutMS</code> is set to value > 0, the function waits up to the given timespan for a new message.
     * If <code>timeoutMS</code> is set to `-1`, the function waits indefinitely until a new message is available.
     * If <code>timeoutMS</code> is set to value lower than -1, results in an exception.
     */
    public function pop(timeoutMS:Int = 0):Null<T> {
        var msg:Null<T> = null;

        if (timeoutMS < -1)
            throw "[timeoutMS] must be >= -1";

        if (timeoutMS == 0) {
            #if (cpp||(hl && haxe_ver >= 4)||neko)
                msg = _queue.pop(false);
            #elseif java
                msg = _queue.poll();
            #elseif python
                msg = try _queue.pop() catch(e:Dynamic) null;
            #else
                _queueLock.acquire();
                msg = _queue.pop();
                _queueLock.release();
            #end
        } else {
            Threads.await(function() {
                #if (cpp||(hl && haxe_ver >= 4)||neko)
                    msg = _queue.pop(false);
                #elseif java
                    msg = _queue.poll();
                #elseif python
                    msg = try _queue.pop() catch(e:Dynamic) null;
                #else
                    _queueLock.acquire();
                    msg = _queue.pop();
                    _queueLock.release();
                #end
                return msg != null;
            }, timeoutMS);
        }
        if (msg != null)
            _length--;
        return msg;
    }
    #else
    public function pop():Null<T> {
        _queueLock.acquire();
        var msg = _queue.pop();
        _queueLock.release();
        return msg;
    }
    #end


    /**
     * Skips the quue and adds the given message to the head of the queue.<commit_msg>Fix Queue.length for platforms without thread support<commit_after>/*
 * Copyright (c) 2016-2018 Vegard IT GmbH, https://vegardit.com
 * SPDX-License-Identifier: Apache-2.0
 */
package hx.concurrent.collection;

import hx.concurrent.atomic.AtomicInt;
import hx.concurrent.lock.RLock;
import hx.concurrent.thread.Threads;

/**
 * Unbound thread-safe first-in-first-out message queue.
 *
 * @author <NAME>, Vegard IT GmbH
 */
class Queue<T> {

    #if cpp
        var _queue = new cpp.vm.Deque<T>();
    #elseif (hl && haxe_ver >= 4)
        var _queue = new hl.vm.Deque<T>();
    #elseif neko
        var _queue = new neko.vm.Deque<T>();
    #elseif java
        var _queue = new java.util.concurrent.ConcurrentLinkedDeque<T>();
    #elseif python
        var _queue:Dynamic;
    #else
        var _queue = new List<T>();
        var _queueLock = new RLock();
    #end

    public var length(get, never):Int;
    var _length = new AtomicInt(0);
    inline function get_length():Int return _length;

    public function new() {
        #if python
            python.Syntax.pythonCode('import collections');
            _queue = untyped collections.deque();
        #end
    }


    #if threads
    /**
     * Pop a message from the queue head.
     *
     * By default (with timeoutMS=0) this function is non-blocking, meaning if no message is available in the queue
     * `null` is returned immediately.
     *
     * If <code>timeoutMS</code> is set to value > 0, the function waits up to the given timespan for a new message.
     * If <code>timeoutMS</code> is set to `-1`, the function waits indefinitely until a new message is available.
     * If <code>timeoutMS</code> is set to value lower than -1, results in an exception.
     */
    public function pop(timeoutMS:Int = 0):Null<T> {
        var msg:Null<T> = null;

        if (timeoutMS < -1)
            throw "[timeoutMS] must be >= -1";

        if (timeoutMS == 0) {
            #if (cpp||(hl && haxe_ver >= 4)||neko)
                msg = _queue.pop(false);
            #elseif java
                msg = _queue.poll();
            #elseif python
                msg = try _queue.pop() catch(e:Dynamic) null;
            #else
                _queueLock.acquire();
                msg = _queue.pop();
                _queueLock.release();
            #end
        } else {
            Threads.await(function() {
                #if (cpp||(hl && haxe_ver >= 4)||neko)
                    msg = _queue.pop(false);
                #elseif java
                    msg = _queue.poll();
                #elseif python
                    msg = try _queue.pop() catch(e:Dynamic) null;
                #else
                    _queueLock.acquire();
                    msg = _queue.pop();
                    _queueLock.release();
                #end
                return msg != null;
            }, timeoutMS);
        }
        if (msg != null) _length--;
        return msg;
    }
    #else
    public function pop():Null<T> {
        _queueLock.acquire();
        var msg = _queue.pop();
        if (msg != null) _length--;
        _queueLock.release();
        return msg;
    }
    #end


    /**
     * Skips the quue and adds the given message to the head of the queue.
{'fraction_non_alphanumeric': 0.11751870324189526, 'fraction_numerical': 0.00685785536159601, 'mean_word_length': 1.734128674904133, 'pattern_counts': {'":': 0, '<': 37, '<?xml version=': 0, '>': 45, 'https://': 2, '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': '3853498', 'n_tokens_mistral': 1966, 'n_tokens_neox': 1867, 'n_words': 557}