Amino というライブラリを使ってみた

これ

CBBs - Concurrent Building Blocks
via:http://secure.ddo.jp/~kaku/tdiary/20090710.html

内容

スレッド 1 で入力を Shift_JIS -> UTF-16 変換して、 UTF-16 のデータをキューに入れる。
スレッド 2 でキューから UTF-16 のデータを取り出して、 UTF-16 -> EUC-JP 変換して出力

結果

$ g++ -O2 -lpthread -licuuc -licudata main.cpp && time ./a.out > out
0.37user 0.03system 0:00.40elapsed 99%CPU (0avgtext+0avgdata 0maxresident)k
0inputs+0outputs (0major+8408minor)pagefaults 0swaps

$ g++ -O2 -lpthread -licuuc -licudata main.cpp -DMULTI_THREAD && time ./a.out > out
0.34user 0.07system 0:00.30elapsed 139%CPU (0avgtext+0avgdata 0maxresident)k
0inputs+0outputs (0major+6784minor)pagefaults 0swaps

うーん

なんか 0.1 秒早くなったけど、たぶん使いどころ違うんだろうなー。

ソース

#include <iostream>
#include <unicode/uclean.h>
#include <unicode/ucnv.h>
#include <amino/thread.h>
#include <amino/queue.h>
#include <boost/tuple/tuple.hpp>

const size_t CHUNKSIZE = 4096;

class DecodeOp
{
    std::istream &in_;
    UErrorCode uerr_;
    UConverter &ucnv_;
    int8_t charSize_;
    amino::LockFreeQueue<boost::tuple<UChar*, size_t> >& queue_;

public:
    DecodeOp(std::istream &in, const std::string &charset, amino::LockFreeQueue<boost::tuple<UChar*, size_t> >& queue)
        :   in_(in),
            uerr_(U_ZERO_ERROR), 
            ucnv_(*ucnv_open(charset.c_str(), &uerr_)),
            charSize_(ucnv_getMinCharSize(&ucnv_)),
            queue_(queue)
    {
    }

    ~DecodeOp()
    {
        ucnv_close(&ucnv_);
    }

    void operator()()
    {
        char inBuf[CHUNKSIZE];

        std::streamsize readSize = 0, consumedSize = 0, lastReadSize = 0;

        while (readSize = in_.rdbuf()->sgetn(inBuf, CHUNKSIZE - lastReadSize + consumedSize))
        {
            const char* inBufBegin  = inBuf;
            const char* inBufEnd    = inBufBegin + readSize;

            size_t bufSize          = charSize_ * readSize;
            UChar* buf              = new UChar[bufSize];
            UChar* bufBegin         = buf;
            const UChar* bufEnd     = bufBegin + bufSize;

            ucnv_toUnicode(&ucnv_, &bufBegin, bufEnd, &inBufBegin, inBufEnd, NULL, false, &uerr_);
            consumedSize = inBufBegin - inBuf;

            queue_.enqueue(boost::tuple<UChar*, size_t>(buf, bufBegin - buf));

            lastReadSize = readSize;
        }

        size_t bufSize          = charSize_ * (lastReadSize - consumedSize);
        UChar* buf              = new UChar[bufSize];
        UChar* bufBegin         = buf;
        const UChar* bufEnd     = bufBegin + bufSize;
        ucnv_toUnicode(&ucnv_, &bufBegin, bufEnd, NULL, NULL, NULL, true, &uerr_);
        queue_.enqueue(boost::tuple<UChar*, size_t>(buf, bufEnd - bufBegin));

        queue_.enqueue(boost::tuple<UChar*, size_t>(NULL, 0));
    }
};

class EncodeOp
{
    std::ostream &out_;
    UErrorCode uerr_;
    UConverter &ucnv_;
    int8_t charSize_;
    amino::LockFreeQueue<boost::tuple<UChar*, size_t> >& queue_;

public:
    EncodeOp(std::ostream &out, const std::string &charset, amino::LockFreeQueue<boost::tuple<UChar*, size_t> >& queue)
        :   out_(out),
            uerr_(U_ZERO_ERROR),
            ucnv_(*ucnv_open(charset.c_str(), &uerr_)),
            charSize_(ucnv_getMaxCharSize(&ucnv_)),
            queue_(queue)
    {
    }

    ~EncodeOp()
    {
        ucnv_close(&ucnv_);
    }

    void operator()()
    {
        boost::tuple<UChar*, size_t> bufData;

        while (true)
        {
            while (!queue_.dequeue(bufData)) { }

            UChar* buf      = boost::get<0>(bufData);
            size_t bufSize  = boost::get<1>(bufData);

            if (buf == NULL)
            {
                break;
            }

            const UChar* bufBegin   = buf;
            const UChar* bufEnd     = buf + bufSize;

            size_t writeSize   = bufSize * charSize_;

            char* outBuf = new char[writeSize];

            char* outBufBegin       = outBuf;
            const char* outBufEnd   = outBufBegin + writeSize;

            ucnv_fromUnicode(&ucnv_, &outBufBegin, outBufEnd, &bufBegin, bufEnd, NULL, false, &uerr_);

            out_.write(outBuf, (outBufBegin - outBuf));

            assert(bufEnd == bufBegin);

            delete[] outBuf;
            delete[] buf;
        }
    }
};

#include <sstream>

int main()
{
    struct Guard {
        UErrorCode err;
        Guard() : err(U_ZERO_ERROR) {
            u_init(&err);
        }
        ~Guard() {
            u_cleanup();
        }
    } guard;

    amino::LockFreeQueue<boost::tuple<UChar*, size_t> > queue;

    std::stringstream s;

    int i = 1000000;
    while (i--)
    {
        s << "\x82\xA0";
        s << "\x82\xA1";
        s << "\x82\xA2";
        s << "\x82\xA3";
    }

    DecodeOp fromSjisOp(s, "Shift_JIS", queue);
    EncodeOp toEucOp(std::cout, "EUC-JP", queue);

#ifdef MULTI_THREAD
    amino::Thread encoderThread(fromSjisOp), decoderThread(toEucOp);
    encoderThread.join();
    decoderThread.join();
#else
    fromSjisOp();
    toEucOp();
#endif

}