// Written in the D programming language.
/*
NYSL Version 0.9982

A. This software is "Everyone'sWare". It means:
  Anybody who has this software can use it as if he/she is
  the author.

  A-1. Freeware. No fee is required.
  A-2. You can freely redistribute this software.
  A-3. You can freely modify this software. And the source
      may be used in any software with no limitation.
  A-4. When you release a modified version to public, you
      must publish it with your name.

B. The author is not responsible for any kind of damages or loss
  while using or misusing this software, which is distributed
  "AS IS". No warranty of any kind is expressed or implied.
  You use AT YOUR OWN RISK.

C. Copyrighted to Kazuki KOMATSU

D. Above three clauses are applied both to source and binary
  form of this software.
*/

/**
このモジュールでは、$(D_CODE std.concurrency)とは違うスレッド間通信を提供します。

このモジュールで提供するスレッド間通信は指定した型のみを送受信できるようなものです。
送受信ともに非同期に行われます。
*/
module carbon.channel;

import carbon.templates;

import std.typetuple;

import lock_free.dlist;


/**
スレッド間通信を提供します。
たとえば、スレッド間でデータを転送する場合に使用します。
*/
shared struct Channel(T...)
{
    static shared(Channel!T) opCall()
    {
        typeof(return) dst;
        foreach(i, E; T)
            dst._chs[i] = new shared AtomicDList!E();

        return dst;
    }


    shared(Sender) sender() @property
    {
        return shared(Sender)(this);
    }


    shared(Receiver) receiver() @property
    {
        return shared(Receiver)(this);
    }


    static shared struct Sender
    {
        void put(U)(U v)
        if(is(U : shared(U)) && isOneOfT!(CastOffShared!U, T))
        {
            _ch.put(v);
        }


      private:
        shared(Channel!T) _ch;
    }


    static shared struct Receiver
    {
        shared(U)* pop(U)() @property
        if(isOneOfT!(U, T))
        {
            return _ch.pop!U;
        }


        shared(AtomicDList!U) queue(U)() @property
        if(isOneOfT!(U, T))
        {
            return _ch.queue!U;
        }

      private:
        shared(Channel!T) _ch;
    }


    void put(U)(U v)
    if(is(U : shared(U)) && isOneOfT!(CastOffShared!U, T))
    {
        foreach(i, E; T){
          static if(is(U == E))
            _chs[i].pushBack(v);
        }
    }


    shared(U)* pop(U)() @property
    if(isOneOfT!(U, T))
    {
        foreach(i, E; T){
          static if(is(U == E))
            return _chs[i].popFront();
        }

        assert(0);
    }


    /// return a queue managed this channel
    shared(AtomicDList!U) queue(U)() @property
    if(isOneOfT!(U, T))
    {
        foreach(i, E; T){
          static if(is(U == E))
            return _chs[i];
        }

        assert(0);
    }


    ///// Wait until appending new element. This function is implemented by core.sync.Condition
    //void waitNewElement()
    //{
    //    _cnd.wait();
    //}


    ///// if this channel is empty, wait until appending new element. This function is implemented by core.sync.Condtion
    //void waitOneElement()
    //{
    //    if(_chs.empty) waitNewElement();
    //}


  private:
    AtomicDLists _chs;
    //Condition _cnd;

    alias AtomicDLists = ToTuple!(TRMap!(AtomicDList, ToTRange!T));
}


shared struct NChannel(size_t N, T...)
{
    static shared(NChannel!(N, T)) opCall()
    {
        typeof(return) dst;

        foreach(i, E; T){
            foreach(j; 0 .. N)
                _chs[i][j] = new shared AtomicDList!E;
        }

        return dst;
    }


    shared(Node!i) node(size_t i)() @property
    {
        return shared(Node!i)(this);
    }


    static shared struct Node(size_t i)
    if(i < N)
    {
        void put(size_t j, U)(U v)
        if(is(U : shared(U)) && isOneOfT!(CastOffShared!U, T) && i != j)
        {
            _ch.put!j(v);
        }


        shared(U)* pop(U)() @property
        if(isOneOfT!(U, T))
        {
            return _ch.pop!i();
        }


        shared(AtomicDList!U) queue(U)() @property
        if(isOneOfT!(U, T))
        {
            return _ch.queue!(i, U)();
        }


      private:
        shared(NChannel!(N, T)) _ch;
    }


    void put(size_t i, U)(U v)
    if(is(U : shared(U)) && isOneOfT!(CastOffShared!U, T) && isOneOfT!(U, T))
    {
        foreach(j, E; T){
          static if(is(U == E))
            _chs[j][i].pushBack(v);
        }
    }


    shared(U)* pop(size_t i, U)(shared U v)
    if(isOneOfT!(U, T))
    {
        foreach(j, E; T){
          static if(is(U == E))
            return _chs[j][i].popFront();
        }
        assert(0);
    }


    shared(AtomicDList!U) queue(size_t i, U)() @property
    if(isOneOfT!(U, T))
    {
        foreach(j, E; T){
          static if(is(U == E))
            return _chs[j][i];
        }
    }

  private:
    AtomicDLists _chs;

    alias AtomicDListN(T) = AtomicDList!T[N];
    alias AtomicDLists = ToTuple!(TRMap!(AtomicDListN, ToTRange!T));

    alias isOneOfT = Channel!(int).isOneOfT;
}


/**
*/
shared(Channel!T) channel(T...)() @property
{
    return shared(Channel!T)();
}


///
shared(NChannel!(N, T)) nchannel(size_t N, T...)() @property
{
    return shared(NChannel!(N, T))();
}

///
unittest{
    import std.concurrency;
    import std.conv;

    auto ch1 = channel!(int, long, string);
    auto ch2 = channel!(int, long, string);

    static void spawnFuncSender(shared ch1.Sender s)
    {
        foreach(i; 0 .. 100){
            s.put(cast(int)i);
            s.put(i + (long.max >> 1) + 1);
            s.put(to!string(i));
        }
    }


    static void spawnFuncMiddle(shared ch1.Receiver r,
                                shared ch2.Sender s)
    {
        foreach(i; 0 .. 100){
            foreach(E; TypeTuple!(int, long, string))
            {
                while(r.queue!E.empty){}
                s.put!E(*r.pop!E);
            }
        }
    }

    spawn(&spawnFuncSender, ch1.sender);
    spawn(&spawnFuncMiddle, ch1.receiver,
                            ch2.sender);

    foreach(i; 0 .. 100){
        foreach(E; TypeTuple!(int, long, string))
        {
            while(ch2.queue!E.empty){}

          static if(is(E == int))
            assert(*ch2.pop!E == i);
          else static if(is(E == long))
            assert(*ch2.pop!E == i + (long.max >> 1) + 1);
          else
            assert(*ch2.pop!E == i.to!string);
        }
    }
}

unittest{
    import std.variant;

    alias Msg = Algebraic!(int, string);
    alias Ch = shared(Channel!Msg);

    auto ch = Ch();
}


private:
template isOneOfT(A, T...)
{
  static if(T.length == 0)
    enum bool isOneOfT = false;
  else
    enum bool isOneOfT = is(A == T[0]) || isOneOfT!(A, T[1 .. $]);
}


template CastOffShared(T)
{
  static if(is(T == shared(U), U))
    alias CastOffShared = U;
  else
    alias CastOffShared = T;
}