前回の記事 : http://qiita.com/muzudho1/items/4de318c07521718857bb
バックグラウンドで動く浮かむ瀬はできたのだった。
次は PHP と 浮かむ瀬 をプロセス間通信させたい。
PHP でのプロセス間通信のやり方は 前に書いたのだった。
http://qiita.com/muzudho1/items/5a36ecdcd9b0ba444577
送る方
<?php
require_once __DIR__ . '/vendor/autoload.php';
use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;
// GETクエリ文字列を取得
$QMsg = urldecode($_SERVER['QUERY_STRING']);
if( "" === $QMsg )
{
$QMsg = "hello, world";
}
// 別のクエリーも送れることを説明
echo '（＾ｑ＾）' . $QMsg . '<br />';
echo 'これを１回クリックしろだぜ☆ｍ９（＾～＾）！<br />';
echo '<a href="http://★Ｗｅｂサイト/tamesi14.php?ClickSitanDaze">http://★Ｗｅｂサイト/tamesi14.php?ClickSitanDaze</a><br />';
// キューを開く
$connection = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');
$channel = $connection->channel();
$QKey = '1111';
// queue_declare(
// $queue = '',
// $passive = false,
// $durable = false,
// $exclusive = false,
// $auto_delete = true,
// $nowait = false,
// $arguments = null,
// $ticket = null
$channel->queue_declare($QKey, false, false, false, false);
// エンキューする
$msgObj = new AMQPMessage( $QMsg );
$channel->basic_publish($msgObj, '', $QKey);
echo "対局プログラムの近くに届けておくぜ☆（＾▽＾）ｖ ['" . $QMsg . "']\n";
// リソースを開放する
$channel->close();
$connection->close();
受け取る方
<?php
// デキュー用
// サーバーの外部からはアクセスできないようにすること。
//
// このプログラムは、コマンドラインで実行する
// php loop6.php
// あるいは
// nohup /usr/bin/php /home/★user/★project/php_service/loop6.php > /home/★user/★project/php_log/loop6.out.log 2> /home/★user/★project/php_log/loop6.err.log < /dev/null &
require_once __DIR__ . '/vendor/autoload.php';
use PhpAmqpLib\Connection\AMQPStreamConnection;
// プロセス間通信の前準備をするぜ☆（＾～＾）
$connection = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');
$channel = $connection->channel();
$QKey = '1111';
$channel->queue_declare($QKey, false, false, false, false);
// キューを開いてしまえば、あとは無限ループ！
// デキューしたタイミングですることをここに書く
$callback = function($msg) {
echo " [x] Received ", $msg->body, "\n";
};
$channel->basic_consume($QKey, '', false, true, false, false, $callback);
// 無限ループ
while(count($channel->callbacks))
{
// ここでブロックしながらデキューする
$channel->wait();
}
// ここに来ない
// $channel->close();
// $connection->close();
無限ループからどうやって抜けんの？
「RabbitMQにPHPクライアント(php-amqplib)からメッセージを送信、受信する」（Symfoware）
http://symfoware.blog68.fc2.com/blog-entry-1490.html
うーむ。
「How on PHP cancel consumer in RabbitMQ?」（stack overflow）
http://stackoverflow.com/questions/24017760/how-on-php-cancel-consumer-in-rabbitmq
「PHP RabbitMQ setTimeout or other option to stop waiting for queue」（stack overflow）
http://stackoverflow.com/questions/33930923/php-rabbitmq-settimeout-or-other-option-to-stop-waiting-for-queue
また usi が効かない
終了するときは[Ctrl]+[C]キーを押せだぜ☆（＾～＾）
エンキューするときはメッセージを打ち込んで[Enter]キーを押せだぜ☆(＾◇＾)
ex.) hello, world
Enqueue? > usi
04:17:55 594 34_cs (P)___ enqueue begin [usi]
04:17:55 594 34_cs (P)___ channel.open begin
04:17:55 594 34_cs (P)___ connection.open begin
04:17:55 607 34_cs (P)___ connection.open end
04:17:55 609 34_cs (P)___ channel.open end
04:17:55 615 34_cs (P)___ enqueue end
04:17:55 615 34_cs (P)___ connection.close begin
04:17:55 615 34_cs (P)___ channel.close begin
04:17:55 622 34_cs (P)___ channel.close end
04:17:55 627 34_cs (P)___ connection.close end
Enqueue? > usi
04:18:05 465 34_cs (P)___ enqueue begin [usi]
04:18:05 465 34_cs (P)___ channel.open begin
04:18:05 465 34_cs (P)___ connection.open begin
04:18:05 475 34_cs (P)___ connection.open end
04:18:05 478 34_cs (P)___ channel.open end
04:18:05 478 34_cs (P)___ enqueue end
04:18:05 478 34_cs (P)___ connection.close begin
04:18:05 478 34_cs (P)___ channel.close begin
04:18:05 480 34_cs (P)___ channel.close end
04:18:05 480 34_cs (P)___ connection.close end
プロセスを調べるか。
# ps aux | grep php
root 6091 0.0 0.0 311040 324 ? Ss Mar07 4:28 php-fpm: master process (/etc/php/7.0/fpm/php-fpm.conf)
www-data 6094 0.0 0.0 311040 0 ? S Mar07 0:00 php-fpm: pool www
www-data 6095 0.0 0.0 311040 0 ? S Mar07 0:00 php-fpm: pool www
root 19662 0.0 0.0 12936 988 pts/0 S+ 04:20 0:00 grep --color=auto php
これが PHP の受信ループだったりしないんだろうか？
# kill 6095
# kill 6094
終了するときは[Ctrl]+[C]キーを押せだぜ☆（＾～＾）
エンキューするときはメッセージを打ち込んで[Enter]キーを押せだぜ☆(＾◇＾)
ex.) hello, world
Enqueue? > usi
04:21:47 062 34_cs (P)___ enqueue begin [usi]
04:21:47 062 34_cs (P)___ channel.open begin
04:21:47 062 34_cs (P)___ connection.open begin
04:21:47 071 34_cs (P)___ connection.open end
04:21:47 072 34_cs (P)___ channel.open end
04:21:47 077 34_cs (P)___ enqueue end
04:21:47 077 34_cs (P)___ connection.close begin
04:21:47 077 34_cs (P)___ channel.close begin
04:21:47 081 34_cs (P)___ channel.close end
04:21:47 087 34_cs (P)___ connection.close end
Enqueue? > 04:21:47 095 34_cs <-------- [interrupt!] Dequeue(^q^) id name ukamuse_SDT4
id author Hiraoka Takuya
option name Best_Book_Move type check default false
option name Book_File type string default book/20150503/book.bin
option name Byoyomi_Margin type spin default 500 min 0 max 2147483647
option name Clear_Hash type button
option name Draw_Ply type spin default 256 min 1 max 2147483647
option name Engine_Name type string default ukamuse_SDT4
option name Max_Book_Ply type spin default 32767 min 0 max 32767
option name Max_Random_Score_Diff type spin default 0 min 0 max 32600
option name Max_Random_Score_Diff_Ply type spin default 32767 min 0 max 32767
option name Min_Book_Ply type spin default 32767 min 0 max 32767
option name Min_Book_Score type spin default -180 min -32601 max 32601
option name Minimum_Thinking_Time type spin default 20 min 0 max 2147483647
option name Move_Overhead type spin default 30 min 0 max 5000
option name MultiPV type spin default 1 min 1 max 594
option name OwnBook type check default true
option name Slow_Mover type spin default 89 min 1 max 1000
option name Slow_Mover_10 type spin default 10 min 1 max 1000
option name Slow_Mover_16 type spin default 20 min 1 max 1000
option name Slow_Mover_20 type spin default 40 min 1 max 1000
option name Threads type spin default 2 min 1 max 256
option name Time_Margin type spin default 4500 min 0 max 2147483647
option name USI_Hash type spin default 256 min 1 max 1048576
option name USI_Ponder type check default true
usiok
あっ、効いた。
isready
position や go を送っても反応がなくなったときは、
isready を送ってみると 反応が返ってくることがある。
初回より後の isready は早い。
position、go 以外に何かステップがあっただろうか？
> isready
< readyok
> usinewgame
> position ～
> go
のサイクルだと回る。なんで isready が要るのかは分からない。
セマフォを作ろう
（セマフォ (semaphore)）「「分かりそう」で「分からない」でも「分かった」気になれるIT用語辞典」
http://wa3.i-3-i.info/word13357.html
ブラウザで多人数同時アクセスが起こるわけだが、とりあえず段階的に
- 浮かむ瀬にアクセスできるのは 同時に１人
- それ以外の人は エラーで 追い払われる
という単純な実装を目指そう。
現状では
- 浮かむ瀬にアクセスできる人数に上限は特にない。RabbitMQがくたばるまで
- アクセスしてもブラウザは待機中になる
- 指し手が返ってきても、それが自分の局面の指し手か分からない（混線）
といった感じになっている。
どうやってセマフォを作るのか？
RabbitMQを用いたプロセス間通信が 逆に弱点になるケースとして、ロック／アンロック 形式は向いていない。
なにしろ、キュー なのだ。キューは「しばらくお待ちください」という原理だ。
強いて言えば ３人並んでいるからまだ大丈夫、と１００人ぐらいが思っていて同時にアクセスして １０３人の行列ができる、というのが 工夫の要らない使い方だ。
「RabbitMQ Tutorials」（RabbitMQ）
http://www.rabbitmq.com/getstarted.html
上のサイトの図を見て分かるように、「Ｐ」は１つで「Ｃ」は１～複数だ。赤いチューブは 待ち行列で、「Ｐ」はいくつでもメッセージを送ってください、という思想だ。「Ｐ」を追い返す思想は 向いていない。
だから、ロック／アンロックを用いて 入場数制限をするための セマフォ は別途用意する必要がある。
とりあえず同時アクセスは そんなに起こらないだろう、という想定にする
RabbitMQ で、キューに溜まっているメッセージ数を調べる方法はあるだろうか。
また、キューに溜まっているのはメッセージ数であって、接続者すうではない。１手指すために isready、usinewgame、position、go の４メッセージを蓄えることになる。ホントは４メッセージも要らないんだが動きが怪しかったのでこの４メッセージを使っている。
ところで PHP側のソースコード
貼り付け忘れていたので。
<?php
require_once __DIR__ . '/vendor/autoload.php';
use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;
//------------------------------------------------------------------------------
// URLの例
//------------------------------------------------------------------------------
//
// http://★サイト/tamesi26a4.php?sfen lnsgkgsnl/1r5b1/ppppppppp/9/9/9/PPPPPPPPP/1B5R1/LNSGKGSNL b - 1 moves
//
// GETクエリ文字列を取得
$QMsg = urldecode($_SERVER['QUERY_STRING']);
if( "" === $QMsg )
{
// 指定が無ければ平手初期局面、先手
$QMsg = "sfen lnsgkgsnl/1r5b1/ppppppppp/9/9/9/PPPPPPPPP/1B5R1/LNSGKGSNL b - 1 moves";
}
//------------------------------------------------------------------------------
// > isready
//------------------------------------------------------------------------------
// キューを選ぶ
// $connection 接続
// $channel チャンネル
// $QName キュー名
// queue_declare(
// $queue = '' , キュー名
// $passive = false, キューの存在確認をするだけ
// $durable = false, ブローカーが停止してもキューは残す
// $exclusive = false, この接続の間だけメッセージを残す
// $auto_delete = true , コンシューマがいなくなるとメッセージを消す
// $nowait = false,
// $arguments = null ,
// $ticket = null
// )
$connection = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');
$channel = $connection->channel();
$QName = '1112';
$channel->queue_declare($QName, false, true, false, true);
// メッセージをエンキューする
$msgObj = new AMQPMessage( 'isready' . "\n" );
$channel->basic_publish($msgObj, '', $QName);
// リソースを開放する
$channel ->close();
$connection->close();
//------------------------------------------------------------------------------
// < readyok
//------------------------------------------------------------------------------
// 接続は使いまわさず、やり直す。
// queue_declare(
// $queue = '' ,
// $passive = false,
// $durable = false,
// $exclusive = false,
// $auto_delete = true ,
// $nowait = false,
// $arguments = null ,
// $ticket = null
// )
$connection = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');
$channel = $connection->channel();
$QName = '1113';
$channel->queue_declare($QName, false, true, false, false);
$callback = function($msg) {
// メッセージをデキューしたタイミングですることをここに書く
echo $msg->body . "\n";
};
$channel->basic_consume($QName, '', false, true, false, false, $callback);
// ここで、メッセージをデキューするまでブロック
$channel->wait();
$channel ->close();
$connection->close();
//------------------------------------------------------------------------------
// > usinewgame
//------------------------------------------------------------------------------
// あとはメッセージの送受信の繰り返し
$connection = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');
$channel = $connection->channel();
$QName = '1112';
$channel->queue_declare($QName, false, true, false, true);
$msgObj = new AMQPMessage( 'usinewgame' . "\n" );
$channel->basic_publish($msgObj, '', $QName);
$channel ->close();
$connection->close();
//------------------------------------------------------------------------------
// > position ～
//------------------------------------------------------------------------------
$connection = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');
$channel = $connection->channel();
$QName = '1112';
$channel->queue_declare($QName, false, true, false, true);
$msgObj = new AMQPMessage( 'position ' . $QMsg . "\n" );
$channel->basic_publish($msgObj, '', $QName);
$channel ->close();
$connection->close();
//------------------------------------------------------------------------------
// > go
//------------------------------------------------------------------------------
$connection = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');
$channel = $connection->channel();
$QName = '1112';
$channel->queue_declare($QName, false, true, false, true);
$msgObj = new AMQPMessage( 'go' . "\n" );
$channel->basic_publish($msgObj, '', $QName);
$channel ->close();
$connection->close();
//------------------------------------------------------------------------------
// < bestmove ～
//------------------------------------------------------------------------------
$connection = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');
$channel = $connection->channel();
$QName = '1113';
$channel->queue_declare($QName, false, true, false, false);
$callback = function($msg) {
echo $msg->body . "\n";
};
$channel->basic_consume($QName, '', false, true, false, false, $callback);
$channel->wait();
$channel->close();
$connection->close();
「php-amqplib」（php-amqplib/php-amqplib）
https://github.com/php-amqplib/php-amqplib
ここから 欲しいメソッドを探すのがむずかしい。
クライアント側の関数から 何件のメッセージが溜まっているか 見えないんじゃないか？
仮に PHP のAPIではキューのメッセージ件数が見れないとしたら、何言語、何スクリプトだったら見えるんだ？
「[C#][PHP][C++] AMQP-CPP の使い方（RabbitMQ、プロセス間通信、メッセージキュー）」（Qiita）
http://qiita.com/muzudho1/items/5a36ecdcd9b0ba444577
C++なら channel.declareQueue のonSuccessハンドラの引数でメッセージ数は見れるようだ。
phpからこれを expect で叩く仕組みを作ればいいだろうか？
じゃあひとまず C++ で作ってみよう。
作った
Makefile
.PHONY: all clean
all: tamesi37a1.exe
tamesi37a1_msgq.o: tamesi37a1_msgq.cpp tamesi37a1_msgq.hpp
g++ -std=c++11 -c tamesi37a1_msgq.cpp -o tamesi37a1_msgq.o -lev -lamqpcpp -pthread
tamesi37a1.o: tamesi37a1.cpp tamesi37a1_msgq.hpp
g++ -std=c++11 -c tamesi37a1.cpp -o tamesi37a1.o -lev -lamqpcpp -pthread
tamesi37a1.exe: tamesi37a1.o tamesi37a1_msgq.o
g++ -std=c++11 tamesi37a1.o tamesi37a1_msgq.o -o tamesi37a1.exe -lev -lamqpcpp -pthread
clean:
rm -f tamesi37a1.d tamesi37a1.o tamesi37a1_msgq.o tamesi37a1.exe tamesi37a1.out.log tamesi37a1.err.log
tamesi37a1_msgq.hpp
// This program for Ubuntu 16.04.
#ifndef TAMESI37A1_MSGQ_HPP
#define TAMESI37A1_MSGQ_HPP
// #pragma once
#include <string>
#include <mutex>
// プロセス間通信用
#include <ev.h>
#include <amqpcpp.h>
#include <amqpcpp/libev.h>
namespace msgq
{
static AMQP::Address ADDRESS{ "amqp://localhost:5672" };
static std::string exchange_name = "myexchange";
static std::string routing_key = "";
// AMQP-CPPでの実装 :
// AMQP::durable=[1] RabbitMQが止まってもキューを残す
// AMQP::autodelete=[2] コンシューマーが１人も接続していなかったら消す
// AMQP::passive=[8] キューが存在するかどうかチェックするだけ。中身見ない時これ
// AMQP::exclusive=[512] この接続でだけ使える。この接続が切れたら消す
typedef int LifeSpan_t;
class Settings
{
std::string name_queue_;
std::mutex _mutex_queueName;
msgq::LifeSpan_t lifeSpan_queue_;
std::mutex _mutex_lifeSpan;
public:
Settings();
~Settings();
void setQueueName(std::string value);
std::string getQueueName();
void setQueueLifeSpan(msgq::LifeSpan_t value);
msgq::LifeSpan_t getQueueLifeSpan();
bool parseCommandlineArgs(int argc, char* argv[]);
std::string Dump();
};
// 接続はシングルトンにします
static struct ev_loop* pLoop_ev = ev_loop_new();
static AMQP::LibEvHandler* pHandler_ev = new AMQP::LibEvHandler(pLoop_ev);
static AMQP::TcpConnection* pConnection_ev = nullptr;
static AMQP::TcpConnection* getConnection()
{
if (nullptr == pConnection_ev) {
pConnection_ev = new AMQP::TcpConnection(pHandler_ev, ADDRESS);
}
return pConnection_ev;
}
static void closeChannel();
static void closeConnection()
{
if (nullptr != pConnection_ev) {
// チャンネルにもヌルのフラグを入れる
closeChannel();
pConnection_ev->close();
pConnection_ev = nullptr;
}
}
// チャンネルはシングルトンにします。
static AMQP::TcpChannel* pChannel_ev = nullptr;
static AMQP::TcpChannel* getChannel()
{
if (nullptr == pChannel_ev)
{
pChannel_ev = new AMQP::TcpChannel(getConnection());
}
return pChannel_ev;
}
static void closeChannel()
{
if (nullptr != pChannel_ev) {
pChannel_ev->close();
pChannel_ev = nullptr;
}
}
}
#endif // #ifndef TAMESI37A1_MSGQ_HPP
tamesi37a1_msgq.cpp
// This program for Ubuntu 16.04.
// #include "stdafx.h"
#include "tamesi37a1_msgq.hpp"
#include <iostream> // std::cout
#include <sstream> // std::ostringstream
namespace msgq
{
static LifeSpan_t LifeSpanString_To_Int(std::string lifeSpan)
{
if ("durable" == lifeSpan) {
return AMQP::durable;
}
else if ("autodelete" == lifeSpan) {
return AMQP::autodelete;
}
else if ("passive" == lifeSpan) {
return AMQP::passive;
}
else if ("exclusive" == lifeSpan) {
return AMQP::exclusive;
}
else {
std::cerr << "未対応のlifeSpan [" << lifeSpan << "]";
exit(11);
}
}
Settings::Settings()
{
name_queue_ = "";
lifeSpan_queue_ = (LifeSpan_t)0;
}
Settings::~Settings()
{
}
void Settings::setQueueName(std::string value) {
std::lock_guard<std::mutex> lock(_mutex_queueName);
name_queue_ = value;
}
std::string Settings::getQueueName() {
std::lock_guard<std::mutex> lock(_mutex_queueName);
return name_queue_;
}
void Settings::setQueueLifeSpan(LifeSpan_t value) {
std::lock_guard<std::mutex> lock(_mutex_lifeSpan);
lifeSpan_queue_ = value;
}
LifeSpan_t Settings::getQueueLifeSpan() {
std::lock_guard<std::mutex> lock(_mutex_lifeSpan);
return lifeSpan_queue_;
}
bool Settings::parseCommandlineArgs(int argc, char* argv[])
{
std::string name_queue = "";
LifeSpan_t lifeSpan_queue = 0;
if (0<argc)
{
// プログラム名を省き、コマンドライン引数だけをつなげる。
std::string cmdArg;
for (int i = 1; i < argc; ++i)
{
cmdArg += std::string(argv[i]);
if (i < argc) {
cmdArg += " ";
}
}
std::istringstream data(cmdArg);
// 与件
// 「--queue 1112 durable autodelete」
// 寿命は可変個数設定可能「durable」「autodelete」「passive」「exclusive」
// 受け皿
// std::string name_queue = "";
// LifeSpan_t lifeSpan_queue = 0;
// 記憶
int m0 = -1; // queue index.
int m1 = -1; // lifespans index.
// 解析器
std::string a;
while (data >> a) {
if ("--" == a.substr(0, 2)) {//先頭の２文字が「--」の場合
if ("--queue" == a) { m0 = 0; m1 = -1; }
else { break; }
}
else if (
">" == a.substr(0, 1)//先頭の１文字が「>」の場合
|| "<" == a.substr(0, 1)//「<」
|| ">" == a.substr(1, 1)//先頭から２文字目が「>」の場合
) {
// 標準入力、標準出力、標準エラーを無視
break;
}
else if (0 == m0) { // キュー名
if (-1 == m1) {
name_queue = a;
lifeSpan_queue = 0;
m1++;
}
else {
lifeSpan_queue |= LifeSpanString_To_Int(a);
}
}
else
{
// 欲しい内容がくるまでスルー
}
}
}
if ("" != name_queue) {
setQueueName(name_queue);
setQueueLifeSpan(lifeSpan_queue);
return true;
}
std::cerr << "コマンドライン引数の「--queue 送信先キュー名 寿命」を漏れなく指定してください。" << std::endl
<< "例: --queue 1113 durable autodelete" << std::endl
<< "寿命は可変個数設定可能「durable」「autodelete」「passive」「exclusive」" << std::endl
<< "name_queue=[" << name_queue << "]" << std::endl;
return false;
}
std::string Settings::Dump()
{
static std::ostringstream sb;
sb << "Dump" << std::endl
<< " name_queue =[" << getQueueName() << "]" << std::endl
<< " lifeSpan_queue =[" << getQueueLifeSpan() << "]" << std::endl
<< " ----" << std::endl;
return sb.str();
}
}
tamesi37a1.cpp
//--------------------------------------------------------------------------------
// OS : Windows10 : It not work. This program for Ubuntu 16.04.
//
// OS : Ubuntu 16.04
// Library : libev
// : Install : Command : sudo apt-get update
// : sudo apt-get install libev-dev
// Service : RabbitMQ
// : Install : Web site : Installing on Debian / Ubuntu http://www.rabbitmq.com/install-debian.html
// : Reference : Web site : Top page http://www.rabbitmq.com/
// : Web site : RabbitMQ管理コマンド（rabbitmqctl）使い方 (Qiita) http://qiita.com/tamikura@github/items/5293cda4c0026b2d7022
// : Web site : amqpを使ってRabbitMQのキューを操作する (Qiita) http://qiita.com/tamikura@github/items/a268afa51c5537ca4fe6
// : Manual : Command : man rabbitmqctl
// : Start : Command : (1) rabbitmq-server
// : (2) [Ctrl] + [Z]
// : (3) bg (Job Number)
// : Stop : Command : rabbitmqctl stop
// : Check : Command : rabbitmqctl status
// : : Command : rabbitmqctl list_queues
// : // Count messages in queues.
// Library : AMQP-CPP
//
// Program : this
// : Author : Satoshi TAKAHASHI (Handle. Muzudho)
// : License : MIT License https://opensource.org/licenses/MIT
// : Explain : This program count messages in queue.
// : Please settings queue to command line argument.
// : Compile : Command : make
// : Explain : Please, read Makefile.
// : Execute : Command : ./tamesi37a1.exe --queue 1112 durable autodelete
// : Run on the foreground.
// : Explain : ./tamesi37a1.exe Executable file
// : --queue queue settings section
// : 1112 queue name (string ok)
// : durable queue life span (multiple)
// : autodelete queue life span
// : Explain : command line argument of life span of queue (Compositable)
// : durable The queue exists, Even if the RabbitMQ service closed.
// : autodelete The queue delete when consumers to be nothing.
// : passive Just check what queue exists. Don't use read/write to queue.
// : exclusive The queue delete when connection closed. The queue exists is this connection only.
// : FAQ : Failures case: Message conflict when starting multiple processes.
// : Example : Windows 10 (This program not work) default queue life span. durable
// : Ubuntu 16.04 default queue life span. durable autodelete
// : Stop : Typing : [Ctrl]+[C]
//
// Referrences :
// : AMQP-CPP : Web site : AMQP-CPP README.md https://github.com/CopernicaMarketingSoftware/AMQP-CPP
// : QueueDeclare http://docs.spring.io/spring-amqp-net/docs/1.0.x/api/html/Spring.Messaging.Amqp.Rabbit~Spring.Messaging.Amqp.Rabbit.Connection.CachedModel~QueueDeclare(String,Boolean,Boolean,Boolean,Boolean,Boolean,IDictionary).html
// : EventingBasicConsumer https://www.rabbitmq.com/releases/rabbitmq-dotnet-client/v1.4.0/rabbitmq-dotnet-client-1.4.0-net-2.0-htmldoc/type-RabbitMQ.Client.Events.EventingBasicConsumer.html
// : BasicConsume https://www.rabbitmq.com/releases/rabbitmq-dotnet-client/v1.4.0/rabbitmq-dotnet-client-1.4.0-net-2.0-htmldoc/type-RabbitMQ.Client.IModel.html#method-M:RabbitMQ.Client.IModel.BasicConsume(System.UInt16,System.String,System.Boolean,System.Collections.IDictionary,RabbitMQ.Client.IBasicConsumer)
// : C#でconstな配列を実現する (もっとクールにプログラミング) http://pgnote.net/?p=885
// : AMQP-CPP (docsforge.com) http://docsforge.com/11/AMQP-CPP/docs/latest/namespace-AMQP/class-TcpConnection/
// : This : Web site : ラムダ式(C++11) （cpprefjp - C++日本語リファレンス） https ://cpprefjp.github.io/lang/cpp11/lambda_expressions.html
//
//--------------------------------------------------------------------------------
#include <string> // std::string
#include <iostream> // std::cout
// プロセス間通信用
#include <ev.h>
#include <amqpcpp.h>
#include <amqpcpp/libev.h>
#include "tamesi37a1_msgq.hpp"
int main(int argc, char* argv[])
{
msgq::Settings settings;
// 引数の解析
if (!settings.parseCommandlineArgs(argc, argv))
{
exit(12);
}
std::string qn = settings.getQueueName();
msgq::LifeSpan_t life = settings.getQueueLifeSpan();
AMQP::TcpChannel* pChannel = msgq::getChannel();
//----------------------------------------
// 指定したキューの残りメッセージ数を返す
//----------------------------------------
auto callback = [](const std::string &name, int msgcount, int consumercount) {
std::cout << "queueName = \"" << name << "\"" << std::endl
<< "messages = " << msgcount << std::endl
<< "nconsumers = " << consumercount << std::endl;
// 接続を切って、このイベントハンドラから抜けることで、ev_runループから抜ける
msgq::closeConnection();
};
pChannel->declareQueue(qn,life)
.onSuccess(callback);
//----------------------------------------
// ループ
//----------------------------------------
ev_run(msgq::pLoop_ev, 0);
return 0;
}
tamesi37a1.expect
#!/usr/bin/expect --
# コマンドライン引数の例
# --queue 1112 durable autodelete
# カレント・ディレクトリを設定
cd /home/★user/★project/cpp_service/t37a1.d
# Expectをタイムアウトさせない
set timeout -1
spawn -noecho ./tamesi37a1.exe ${argv}
expect -re "queueName = "
expect -re "messages = "
expect -re "consumers = "
tamesi26a5.php
<?php
// 処理時間計測
$time=microtime(true);
require_once __DIR__ . '/vendor/autoload.php';
use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;
//------------------------------------------------------------------------------
// URLの例
//------------------------------------------------------------------------------
//
// http://★サイト/tamesi26a5.php?sfen lnsgkgsnl/1r5b1/ppppppppp/9/9/9/PPPPPPPPP/1B5R1/LNSGKGSNL b - 1 moves
//
// GETクエリ文字列を取得
$QMsg = urldecode($_SERVER['QUERY_STRING']);
if( '' === $QMsg )
{
$QMsg = 'sfen lnsgkgsnl/1r5b1/ppppppppp/9/9/9/PPPPPPPPP/1B5R1/LNSGKGSNL b - 1 moves';
}
//------------------------------------------------------------------------------
// busy確認
//------------------------------------------------------------------------------
exec('/usr/bin/expect /home/★user/★project/expect_service/tamesi37a1.expect --queue 1112 durable autodelete',$o);
foreach($o as $v)
{
if(0===strpos($v,'messages = '))
{
$messages = intval( substr($v,11) );
if( 0 < $messages ){
die( 'error = "busy"' . "\n" . 'busy = ' . $messages . "\n" );
}
}
//else
//{
// echo 'v = ' . $v . "\n";
//}
}
//------------------------------------------------------------------------------
// > isready
//------------------------------------------------------------------------------
// キューを開く
$connection = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');
$channel = $connection->channel();
$QName = '1112';
// queue_declare(
// $queue = '',
// $passive = false,
// $durable = false,
// $exclusive = false,
// $auto_delete = true,
// $nowait = false,
// $arguments = null,
// $ticket = null
$channel->queue_declare($QName, false, true, false, true);
// エンキューする
$msgObj = new AMQPMessage( 'isready' . "\n" );
$channel->basic_publish($msgObj, '', $QName);
// リソースを開放する
$channel->close();
$connection->close();
//------------------------------------------------------------------------------
// < readyok
//------------------------------------------------------------------------------
// 接続は使いまわさず、やり直す。
$connection = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');
$channel = $connection->channel();
$QName = '1113';
// queue_declare(
// $queue = '',
// $passive = false,
// $durable = false,
// $exclusive = false,
// $auto_delete = true,
// $nowait = false,
// $arguments = null,
// $ticket = null
$channel->queue_declare($QName, false, true, false, false);
// キューを開いてしまえば、あとは無限ループ！
// デキューしたタイミングですることをここに書く
$callback = function($msg) {
if('readyok' !== rtrim($msg->body))
{
die( "error = \"no readyok.\"\n" . $msg->body . "\n" );
}
};
$channel->basic_consume($QName, '', false, true, false, false, $callback);
// ここでブロックしながらデキューする。強制終了するまで続く
// wait($allowed_methods=null, $non_blocking = false, $timeout = 0)
//$timeoutSeconds = 2;
//$channel->wait(null, false, timeoutSeconds);
$channel->wait();
$channel->close();
$connection->close();
//------------------------------------------------------------------------------
// > usinewgame
//------------------------------------------------------------------------------
// キューを開く
$connection = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');
$channel = $connection->channel();
$QName = '1112';
$channel->queue_declare($QName, false, true, false, true);
// エンキューする
$msgObj = new AMQPMessage( 'usinewgame' . "\n" );
$channel->basic_publish($msgObj, '', $QName);
// リソースを開放する
$channel->close();
$connection->close();
//------------------------------------------------------------------------------
// > position ～
//------------------------------------------------------------------------------
// キューを開く
$connection = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');
$channel = $connection->channel();
$QName = '1112';
$channel->queue_declare($QName, false, true, false, true);
// エンキューする
$msgObj = new AMQPMessage( 'position ' . $QMsg . "\n" );
$channel->basic_publish($msgObj, '', $QName);
// リソースを開放する
$channel->close();
$connection->close();
//------------------------------------------------------------------------------
// > go
//------------------------------------------------------------------------------
// キューを開く
$connection = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');
$channel = $connection->channel();
$QName = '1112';
$channel->queue_declare($QName, false, true, false, true);
// エンキューする
$msgObj = new AMQPMessage( 'go' . "\n" );
$channel->basic_publish($msgObj, '', $QName);
// リソースを開放する
$channel->close();
$connection->close();
//------------------------------------------------------------------------------
// < bestmove ～
//------------------------------------------------------------------------------
$connection = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');
$channel = $connection->channel();
$QName = '1113';
$channel->queue_declare($QName, false, true, false, false);
$callback = function($msg) {
echo $msg->body . "\n";
};
$channel->basic_consume($QName, '', false, true, false, false, $callback);
// ここでブロックしながらデキューする。強制終了するまで続く
$channel->wait();
$channel->close();
$connection->close();
//------------------------------------------------------------------------------
// 処理時間表示
//------------------------------------------------------------------------------
echo '#<br /><br />' . "\ntime = " . (microtime(true) - $time);
浮かむ瀬をどのように改造したのか
common.hpp
// (2017-03-22 Modify begin) 標準入出力は全て変更
#define SYNCCOUT SERVIO_BEGIN
#define SYNCENDL SERVIO_END_CLEAN
// (2017-03-22 Modify end)
標準入出力は全て変更した。量が多いので１つ１つは見ていけないが……。
search.cpp
// (2017-03-22 Modify begin) 標準入出力は全て変更
if (nyugyokuWin)
{
SERVIO_BEGIN << "bestmove win" << SERVIO_END_FLUSH(searcher, "NyugyokuWin");
}
else if (!bestThread->rootMoves[0].pv[0])
{
SERVIO_BEGIN << "bestmove resign" << SERVIO_END_FLUSH(searcher, "Resign");
}
else {
if (bestThread->rootMoves[0].pv.size() > 1 || bestThread->rootMoves[0].extractPonderFromTT(pos)) {
// ポンダー付きの場合
SERVIO_BEGIN << "bestmove " << bestThread->rootMoves[0].pv[0].toUSI() << " ponder " << bestThread->rootMoves[0].pv[1].toUSI() << SERVIO_END_FLUSH(searcher, "BestmoveAndPonder");
}
else {
SERVIO_BEGIN << "bestmove " << bestThread->rootMoves[0].pv[0].toUSI() << SERVIO_END_FLUSH(searcher, "Bestmove");
}
}
// (2017-03-22 Modify end)
こんな風に、標準出力ストリームがあるところは 全て置き換えた、あるいは見た目上変わりはないが、マクロの中身が変わったと考えていい。
USIループも変えた
標準出力の改造より、標準入力の改造の方が 大がかりだ。
usi.cpp
void Searcher::doUSICommandLoop(int argc, char* argv[]) {
#if UBUNTU
// 引数の解析
std::string name_enqueue = "";
service_io::LifeSpan_t lifeSpan_enqueue = 0;
std::string name_dequeue = "";
service_io::LifeSpan_t lifeSpan_dequeue = 0;
{
if (!service_io::parseArgs(argc, argv, name_enqueue, lifeSpan_enqueue, name_dequeue, lifeSpan_dequeue))
{
exit(12);
}
thisptr->serviceIo2.SetQueueName (service_io::ENQUEUE_INDEX, name_enqueue);
thisptr->serviceIo2.SetQueueLifeSpan(service_io::ENQUEUE_INDEX, lifeSpan_enqueue);
thisptr->serviceIo2.SetQueueName (service_io::DEQUEUE_INDEX, name_dequeue);
thisptr->serviceIo2.SetQueueLifeSpan(service_io::DEQUEUE_INDEX, lifeSpan_dequeue);
}
// デキューの常時監視をスタート
auto th1 = std::thread([] { service_io::workConsume(*thisptr); });
#endif
bool dialogue = false;
// 「./apery」のように引数無しで打鍵して実行された場合に限り、なんども標準入力を待つ。
if (argc == 1)
{
dialogue = true;
}
bool evalTableIsRead = false;
Position pos(DefaultStartPositionSFEN, threads.main(), thisptr);
std::string cmd;
std::string token;
// プログラム名を省き、コマンドライン引数だけをつなげる。
for (int i = 1; i < argc; ++i)
{
cmd += std::string(argv[i]) + " ";
}
do {
// 文字が打たれたらその文字を、空文字を打たれたら "quit" 扱いとする。
if (dialogue)
{
cmd = service_io::dequeue();
}
// 半角スペース区切りの最初の1トークンを拾う
std::istringstream ssCmd(cmd);
ssCmd >> std::skipws >> token;
if (token == "--enqueue" || token == "--dequeue") {
#if UBUNTU
dialogue = true;
#else
std::cerr << "Command line parameter error : \"--msgqueue\" option for UBUNTU.";
#endif
}
// else if (token == "tu") {
//#if UBUNTU
// // tu と飛んできたら ka と返す
// service_io::oss << "ka";
// service_io::flushMessage();
//#else
// std::cerr << "Command error : \"tu\" command for UBUNTU.";
//#endif
// }
else if (token == "quit" || token == "stop" || token == "ponderhit" || token == "gameover") {
if (token != "ponderhit" || signals.stopOnPonderHit) {
signals.stop = true;
threads.main()->startSearching(true);
}
else
limits.ponder = false;
if (token == "ponderhit" && limits.moveTime != 0)
limits.moveTime += timeManager.elapsed();
}
else if (token == "go") go(pos, ssCmd);
else if (token == "position") setPosition(pos, ssCmd);
else if (token == "usinewgame"); // isready で準備は出来たので、対局開始時に特にする事はない。
else if (token == "usi")
{
SERVIO_BEGIN << "id name " << std::string(options["Engine_Name"])
<< "\nid author Hiraoka Takuya"
<< "\n" << options
<< "\nusiok" << SERVIO_END_FLUSH(thisptr, "Usi");
}
else if (token == "isready") { // 対局開始前の準備。
tt.clear();
threads.main()->previousScore = ScoreInfinite;
if (!evalTableIsRead) {
// 一時オブジェクトを生成して Evaluator::init() を呼んだ直後にオブジェクトを破棄する。
// 評価関数の次元下げをしたデータを格納する分のメモリが無駄な為、
std::unique_ptr<Evaluator>(new Evaluator)->init(Evaluator::evalDir, true);
evalTableIsRead = true;
}
SERVIO_BEGIN << "readyok" << SERVIO_END_FLUSH(thisptr, "Isready");
}
else if (token == "setoption") setOption(ssCmd);
else if (token == "write_eval") { // 対局で使う為の評価関数バイナリをファイルに書き出す。
if (!evalTableIsRead)
std::unique_ptr<Evaluator>(new Evaluator)->init(Evaluator::evalDir, true);
Evaluator::writeSynthesized(Evaluator::evalDir);
}
#if defined LEARN
else if (token == "l") {
auto learner = std::unique_ptr<Learner>(new Learner);
learner->learn(pos, ssCmd);
}
else if (token == "make_teacher") {
if (!evalTableIsRead) {
std::unique_ptr<Evaluator>(new Evaluator)->init(Evaluator::evalDir, true);
evalTableIsRead = true;
}
make_teacher(ssCmd);
}
else if (token == "use_teacher") {
if (!evalTableIsRead) {
std::unique_ptr<Evaluator>(new Evaluator)->init(Evaluator::evalDir, true);
evalTableIsRead = true;
}
use_teacher(pos, ssCmd);
}
else if (token == "check_teacher") {
check_teacher(ssCmd);
}
else if (token == "print") printEvalTable(SQ88, f_gold + SQ78, f_gold, false);
#endif
#if !defined MINIMUL
// 以下、デバッグ用
else if (token == "bench") {
if (!evalTableIsRead) {
std::unique_ptr<Evaluator>(new Evaluator)->init(Evaluator::evalDir, true);
evalTableIsRead = true;
}
benchmark(pos);
}
else if (token == "key") { SYNCCOUT << pos.getKey() << SYNCENDL; }
else if (token == "tosfen") { SYNCCOUT << pos.toSFEN() << SYNCENDL; }
else if (token == "eval") { SYNCCOUT << evaluateUnUseDiff(pos) / FVScale << SYNCENDL; }
else if (token == "d") { pos.print(); }
else if (token == "s") { measureGenerateMoves(pos); }
else if (token == "t") { SYNCCOUT << pos.mateMoveIn1Ply().toCSA() << SYNCENDL; }
else if (token == "b") { makeBook(pos, ssCmd); }
#endif
else { SYNCCOUT << "unknown command: [" << cmd << "]" << SYNCENDL; }
} while (token != "quit" && dialogue);
#if UBUNTU
// th1 スレッドの終了を待つ
th1.join();
#endif
threads.main()->waitForSearchFinished();
}
バックグラウンドに常駐させる前提なので、標準入出力は塞ぎ、通信はメッセージ・キューを用いたプロセス間通信に置き換えた。
メッセージの送信先をエンキュー、
メッセージの受信元をデキュー と呼ぶことにした。
service_io::dequeue();
という部分で ブラウザから送られてくるメッセージを取得している。
また apery はコマンドライン引数を 単にコマンドと解釈してループの中で解析するので、
コマンドライン引数「--enqueue ～略～ --dequeue ～略～」を使いたいときに困った。
そこで 改造を加えたが、これが悪影響を与えていないものかどうか。
グローバル変数 searcher
コンピュータ将棋はマルチスレッドで動いているので、グローバル変数を用意しても、別スレッドからは見えないので グローバルには利用できない。
そこで、どのスレッドからも利用できるグローバル変数のように利用されている Searcher 構造体に、メッセージ・キュー名等の、グローバルに利用したいデータを追加した。
search.hpp
// (2017-03-20 Add begin)
#ifdef UBUNTU
class ServiceIo2
{
public:
std::string name_queues_[2];// = { "", "" };
std::mutex _mutex_queueName;
service_io::LifeSpan_t lifeSpan_queues_[2];// = { (service_io::LifeSpan_t)0, (service_io::LifeSpan_t)0 };
std::mutex _mutex_lifeSpan;
void SetQueueName(int index, std::string value);
std::string GetQueueName(int index);
void SetQueueLifeSpan(int index, service_io::LifeSpan_t value);
service_io::LifeSpan_t GetQueueLifeSpan(int index);
};
#endif
// (2017-03-20 Add end)
構造体の中
// (2017-03-20 Add begin)
#ifdef UBUNTU
public:
STATIC ServiceIo2 serviceIo2;
#endif
// (2017-03-20 Add end)
ファイルの末尾
// (2017-03-20 Add begin)
namespace service_io
{
#ifdef UBUNTU
// メッセージキューにエンキュー
// 出力ストリームのカンマ区切り「,,,;」で使われ、何か数字を返す必要があって必然性はない数字 0 を返している
static int flushMessage(Searcher* pSearcher, std::string hint)
{
std::string message = service_io::oss.str();
oss.str(std::string());// 空文字列を入れてクリアーする
std::string name_enqueue = pSearcher->serviceIo2.GetQueueName(ENQUEUE_INDEX);
LifeSpan_t lifeSpan_enqueue = pSearcher->serviceIo2.GetQueueLifeSpan(ENQUEUE_INDEX);
if ("" == name_enqueue)
{
std::cerr << getTimeslipNow() << "(>_<)ukamuse_sdt4_child4 (P)___ WKP.SM_ queue name is empty=[" << name_enqueue << "] ENQUEUE_INDEX=[" << ENQUEUE_INDEX << "] lifeSpan=[" << lifeSpan_enqueue << "] hint=[" << hint << "]" << "\n";
exit(1);
}
// チャンネルを宣言
getChannel(ENQUEUE_INDEX)->declareQueue(name_enqueue, lifeSpan_enqueue)
.onError([=](const char* errMsg) {
std::cerr << getTimeslipNow() << "(>_<)ukamuse_sdt4_child4 (P)___ WKP.SM_.declareQueue.onError [" << errMsg << "] queue name=[" << name_enqueue << "] lifeSpan=[" << lifeSpan_enqueue << "]" << "\n";
});
// チャンネルを束縛
getChannel(ENQUEUE_INDEX)->bindQueue(exchange_name, name_enqueue, routing_key)
// ev_run から１回だけ呼び出される。コミットを開始してこのハンドラから抜けること。コミットの成功時に接続を切ることでev_runから抜ける。
.onSuccess([=]() {//&message
// トランザクション開始
getChannel(ENQUEUE_INDEX)->startTransaction();
// パブリッシュ開始
if (getChannel(ENQUEUE_INDEX)->publish(exchange_name, routing_key, message.c_str(), message.size())) {
// コミットを開始
getChannel(ENQUEUE_INDEX)->commitTransaction()
.onSuccess([=]() {
// コミット成功
// 接続を切ったあと、トランザクションのコミットが実行されるようだ。
closeConnection(ENQUEUE_INDEX);
})
.onError([=](const char *errMsg) {//&message
std::cerr << getTimeslipNow() << "(>_<)ukamuse_sdt4_child4 (P)___ WKP.BQ_.OS_ CT.onError failuerCommit errMsg=[" << errMsg << "] message=[" << message << "]" << std::endl;
});
}
else {
std::cerr << getTimeslipNow() << "(>_<)ukamuse_sdt4_child4 (P)___ WKP.BQ_.OS_ else failuer message=[" << message << "]" << std::endl;
getChannel(ENQUEUE_INDEX)->rollbackTransaction();
}
}).onError([](const char* errMsg) {
std::cerr << getTimeslipNow() << "(>_<)ukamuse_sdt4_child4 (P)___ WKP.bindQueue.onError [" << errMsg << "]" << std::endl;
});
// ここで、バインドキューのオンサクセスの処理へ。
ev_run(pLoop_ev[ENQUEUE_INDEX], 0);
return 0;
}
// メッセージ・キューの監視を開始
static void workConsume(Searcher& searcher)
{
std::string name_dequeue = searcher.serviceIo2.GetQueueName(DEQUEUE_INDEX);
LifeSpan_t lifeSpan_dequeue = searcher.serviceIo2.GetQueueLifeSpan(DEQUEUE_INDEX);
// I will go to the front of the box named "1111".
getChannel(DEQUEUE_INDEX)->declareQueue(name_dequeue, lifeSpan_dequeue);
// I look inside the box.
auto errorCb = [&lifeSpan_dequeue](const char *errMsg) {
std::cerr << getTimeslipNow() << "(>_<)cpp ___(C) CNS.onError My ID watching failed [" << errMsg << "] lifeSpan_dequeue=[" << lifeSpan_dequeue << "]" << std::endl;
};
auto messageCb = [](const AMQP::Message &amqpMessage, uint64_t deliveryTag, bool redelivered) {
std::string message(amqpMessage.body(), amqpMessage.bodySize());
rotationBuffer::push_block(message);
getChannel(DEQUEUE_INDEX)->ack(deliveryTag);
};
getChannel(DEQUEUE_INDEX)->consume(name_dequeue)
.onReceived(messageCb)
.onError(errorCb);
// I will keep on forever.
ev_run(pLoop_ev[DEQUEUE_INDEX]);
// I will not come here.
return;
}
#else
static int flushMessage(Searcher* pSearcher, std::string hint) {
std::cout << IOLock << service_io::oss.str() << std::endl << IOUnlock;
service_io::oss.str(std::string());// 空文字列を入れてクリアーする
return 0;
}
#endif
}
// (2017-03-20 Add end)
search.cpp
// (2017-03-20 Add begin)
#ifdef UBUNTU
ServiceIo2 Searcher::serviceIo2;
#endif
// (2017-03-20 Add end)
init( )
// (2017-03-20 Add begin)
#ifdef UBUNTU
serviceIo2.name_queues_[service_io::ENQUEUE_INDEX] = "";
serviceIo2.name_queues_[service_io::DEQUEUE_INDEX] = "";
serviceIo2.lifeSpan_queues_[service_io::ENQUEUE_INDEX] = (service_io::LifeSpan_t)0;
serviceIo2.lifeSpan_queues_[service_io::DEQUEUE_INDEX] = (service_io::LifeSpan_t)0;
#endif
// (2017-03-20 Add end)
ファイル末尾
// (2017-03-20 Add begin)
#ifdef UBUNTU
void ServiceIo2::SetQueueName(int index, std::string value) {
std::lock_guard<std::mutex> lock(_mutex_queueName);
name_queues_[index] = value;
}
std::string ServiceIo2::GetQueueName(int index) {
std::lock_guard<std::mutex> lock(_mutex_queueName);
return name_queues_[index];
}
void ServiceIo2::SetQueueLifeSpan(int index, service_io::LifeSpan_t value) {
std::lock_guard<std::mutex> lock(_mutex_lifeSpan);
lifeSpan_queues_[index] = value;
}
service_io::LifeSpan_t ServiceIo2::GetQueueLifeSpan(int index) {
std::lock_guard<std::mutex> lock(_mutex_lifeSpan);
return lifeSpan_queues_[index];
}
#endif
// (2017-03-20 Add end)
また、ファイルも２つ追加した。
service_io.hpp
// OS : Windows 10
// IDE : Visual Studio 2015
// Install : NuGet : Install-Package RabbitMQ.Client -Version 4.1.1
//
// OS : Ubuntu 16.04
// Service : RabbitMQ
// : Install : Web site : Installing on Debian / Ubuntu http://www.rabbitmq.com/install-debian.html
// : Reference : Web site : Top page http://www.rabbitmq.com/
// : Web site : RabbitMQ管理コマンド（rabbitmqctl）使い方 (Qiita) http://qiita.com/tamikura@github/items/5293cda4c0026b2d7022
// : Web site : amqpを使ってRabbitMQのキューを操作する (Qiita) http://qiita.com/tamikura@github/items/a268afa51c5537ca4fe6
// : Manual : Command : man rabbitmqctl
// : Start : Command : (1) rabbitmq-server
// : (2) [Ctrl] + [Z]
// : (3) bg (Job Number)
// : Stop : Command : rabbitmqctl stop
// : Check : Command : rabbitmqctl status
// : Command : rabbitmqctl list_queues
// : // Count messages in queues.
//
// Program : this
// : Compile : Command : make profgen_sse
// : : mv apery ../bin/apery
// : Execute : Command : ./apery --enqueue 1113 durable --dequeue 1112 durable autodelete > ./apery.out.log 2> ./apery.err.log < /dev/null &
// 対局
// ====
// usi
// usiok
// isready
// readyok
// usinewgame
// position sfen lnsgkgsnl/1r5b1/ppppppppp/9/9/9/PPPPPPPPP/1B5R1/LNSGKGSNL b - 1 moves
// go
// 例) bestmove 2g2f ponder 1c1d
// quit
#ifndef APERY_SERVICE_IO_HPP
#define APERY_SERVICE_IO_HPP
#include <string> // std::string
#include <iostream> // std::cout
#include <sstream> // std::ostringstream
#include "common.hpp"
// #define UBUNTU
// 入出力を、サービスIOでラッピングします
#define SERVIO_BEGIN service_io::oss
#define SERVIO_END_CLEAN std::endl, service_io::cleanMessage()
#define SERVIO_END_FLUSH(pSearcher,hint) std::endl, service_io::flushMessage(pSearcher,hint)
namespace service_io
{
// 文字列ストリーム出力を、文字列に置換するもの
static std::ostringstream oss;
}
#ifdef UBUNTU
// OS : Ubuntu 16.04
// Library : libev
// : Install : Command : sudo apt-get update
// : sudo apt-get install libev-dev
// Service : RabbitMQ
// : Reference : Web site : Top page http://www.rabbitmq.com/
// : Install : Web site : Installing on Debian / Ubuntu http://www.rabbitmq.com/install-debian.html
// : Manual : Command : man rabbitmqctl
// : Start : Command : rabbitmq-server
// : Stop : Command : rabbitmqctl stop
// : Check : Command : rabbitmqctl status
// : : Command : rabbitmqctl list_queues
// Library : AMQP-CPP
// : Reference : Web site : AMQP-CPP README.md https://github.com/CopernicaMarketingSoftware/AMQP-CPP
//
// service_io.hpp
// プロセス間通信用
#include <ev.h>
#include <amqpcpp.h>
#include <amqpcpp/libev.h>
// バックグラウンド・プロセスに切り替えて使用する可能性も考慮した入出力API
namespace service_io
{
// 時計はずれているが、現在時刻が返ってくる
static std::string getTimeslipNow()
{
// どこか外国の 1969年12月31日19時ぐらいから数えたミリ秒だろうか？
std::chrono::milliseconds ms = std::chrono::duration_cast< std::chrono::milliseconds >(
std::chrono::system_clock::now().time_since_epoch()
);
long long hours = ms.count() / 1000 / 60 / 60 % 24;
long long minutes = ms.count() / 1000 / 60 % 60;
long long seconds = ms.count() / 1000 % 60;
long long milliseconds = ms.count() % 1000;
std::ostringstream oss;
oss << std::setfill('0') << std::setw(2) << hours << ":" << std::setfill('0') << std::setw(2) << minutes << ":" << std::setfill('0') << std::setw(2) << seconds << "'" << std::setfill('0') << std::setw(3) << milliseconds << " ";
return oss.str();
}
// AMQP-CPPでの実装 :
// AMQP::durable=[1] RabbitMQが止まってもキューを残す
// AMQP::autodelete=[2] コンシューマーが１人も接続していなかったら消す
// AMQP::passive=[8] キューが存在するかどうかチェックするだけ。中身見ない時これ
// AMQP::exclusive=[512] この接続でだけ使える。この接続が切れたら消す
typedef int LifeSpan_t;
// 0 : enqueue用設定
// 1 : dequeue用設定
static const int ENQUEUE_INDEX = 0;
static const int DEQUEUE_INDEX = 1;
static const int NUM_INDEX = 2;
static LifeSpan_t LifeSpanString_To_Int(std::string lifeSpan)
{
if ("durable" == lifeSpan) {
return AMQP::durable;
}
else if ("autodelete" == lifeSpan) {
return AMQP::autodelete;
}
else if ("passive" == lifeSpan) {
return AMQP::passive;
}
else if ("exclusive" == lifeSpan) {
return AMQP::exclusive;
}
else {
std::cerr << "未対応のlifeSpan [" << lifeSpan << "]";
exit(11);
}
}
static AMQP::Address ADDRESS{ "amqp://localhost:5672" };
static std::string exchange_name = "myexchange";
static std::string routing_key = "";
// 接続はシングルトンにします
static struct ev_loop* pLoop_ev[2] = { ev_loop_new(), ev_loop_new() };
static AMQP::LibEvHandler* pHandler_ev[2] = { new AMQP::LibEvHandler(pLoop_ev[ENQUEUE_INDEX]), new AMQP::LibEvHandler(pLoop_ev[DEQUEUE_INDEX]) };
static AMQP::TcpConnection* pConnection_ev[2] = { nullptr,nullptr };
static AMQP::TcpConnection* getConnection(int index)
{
if (nullptr == pConnection_ev[index]) {
pConnection_ev[index] = new AMQP::TcpConnection(pHandler_ev[index], ADDRESS);
}
return pConnection_ev[index];
}
static void closeChannel(int index);
static void closeConnection(int index)
{
if (nullptr != pConnection_ev[index]) {
// チャンネルにもヌルのフラグを入れる
closeChannel(index);
pConnection_ev[index]->close();
pConnection_ev[index] = nullptr;
}
}
// チャンネルはシングルトンにします。
static AMQP::TcpChannel* pChannel_ev[2] = { nullptr,nullptr };
static AMQP::TcpChannel* getChannel(int index)
{
if (nullptr == pChannel_ev[index])
{
pChannel_ev[index] = new AMQP::TcpChannel(getConnection(index));
}
return pChannel_ev[index];
}
static void closeChannel(int index)
{
if (nullptr != pChannel_ev[index]) {
pChannel_ev[index]->close();
pChannel_ev[index] = nullptr;
}
}
/// <summary>
/// 回転式バッファー。
/// これはメイン・スレッドに置く。
/// デキューのスレッドでエンキューすることはできない。
/// デキュー処理は、回転式バッファーを仲介にしてエンキュー処理にメッセージを渡す。
/// </summary>
namespace rotationBuffer
{
const int bufferSize = 1024;
static std::string buffer[bufferSize] = {};
static int bufferCursors[2] = { 0, 0 };
static int busy = 0;
const int PUT_INDEX = 0;
const int GET_INDEX = 1;
static std::mutex _mutex;
// 返り値: 追加しなかったとき 偽、それ以外のとき 真。
static bool push_inner(std::string message)
{
std::lock_guard<std::mutex> lock(_mutex);
if (busy + 1 == bufferSize || bufferSize * 99 / 100 < busy) {
// busy
return false;
}
buffer[bufferCursors[PUT_INDEX]] = message;
bufferCursors[PUT_INDEX]++;
busy++;
if (!(bufferCursors[PUT_INDEX] < bufferSize))
{
bufferCursors[PUT_INDEX] = 0;
}
return true;
}
static std::string pop_inner()
{
std::lock_guard<std::mutex> lock(_mutex);
if ("" != buffer[bufferCursors[GET_INDEX]])
{
std::string message = buffer[bufferCursors[GET_INDEX]];
buffer[bufferCursors[GET_INDEX]] = "";
bufferCursors[GET_INDEX]++;
busy--;
if (!(bufferCursors[GET_INDEX] < bufferSize))
{
bufferCursors[GET_INDEX] = 0;
}
return message;
}
// buffer is empty
return "";
}
// 追加に成功するまで、ブロックする
static void push_block(std::string message)
{
while (!rotationBuffer::push_inner(message))
{
// busy
std::this_thread::sleep_for(std::chrono::milliseconds(40));
}
}
// 何か取れるまで、ブロックする
static std::string pop_block()
{
std::string message;
while ("" == (message = rotationBuffer::pop_inner()))
{
// buffer is empty
std::this_thread::sleep_for(std::chrono::milliseconds(20));
}
return message;
}
}
static int cleanMessage() {
service_io::oss.str(std::string());// 空文字列を入れてクリアーする
return 0;
}
// service_io::workConsume() を別スレッドで走らせておくこと。
static std::string dequeue() {
return rotationBuffer::pop_block();
}
static bool parseArgs(int argc, char* argv[], std::string& name_enqueue, LifeSpan_t& lifeSpan_enqueue, std::string& name_dequeue, LifeSpan_t& lifeSpan_dequeue)
{
name_enqueue = "";
name_dequeue = "";
if (0<argc)
{
// プログラム名を省き、コマンドライン引数だけをつなげる。
std::string cmdArg;
for (int i = 1; i < argc; ++i)
{
cmdArg += std::string(argv[i]);
if (i < argc) {
cmdArg += " ";
}
}
std::istringstream data(cmdArg);
// 与件
// 「--enqueue 1113 durable --dequeue 1112 durable autodelete」
// 寿命は可変個数設定可能「durable」「autodelete」「passive」「exclusive」
// 受け皿
// 参照渡しの引数に返す
// 記憶
int m0 = -1; // enqueue or dequeue index.
int m1 = -1; // lifespans index.
// 解析器
std::string a;
while (data >> a) {
if ("--" == a.substr(0, 2)) {//先頭の２文字が「--」の場合
if ("--enqueue" == a) { m0 = ENQUEUE_INDEX; m1 = -1; }
else if ("--dequeue" == a) { m0 = DEQUEUE_INDEX; m1 = -1; }
else { break; }
}
else if (
">" == a.substr(0, 1)//先頭の１文字が「>」の場合
|| "<" == a.substr(0, 1)//「<」
|| ">" == a.substr(1, 1)//先頭から２文字目が「>」の場合
) {
// 標準入力、標準出力、標準エラーを無視
break;
}
else if (0 == m0) { // エンキュー先キュー名
if (-1 == m1) {
name_enqueue = a;
lifeSpan_enqueue = 0;
m1++;
}
else {
lifeSpan_enqueue |= LifeSpanString_To_Int(a);
}
}
else if (1 == m0) { // デキュー元キュー名
if (-1 == m1) {
name_dequeue = a;
lifeSpan_dequeue = 0;
m1++;
}
else {
lifeSpan_dequeue |= LifeSpanString_To_Int(a);
}
}
else
{
// 欲しい内容がくるまでスルー
}
}
if ("" == name_enqueue || "" == name_dequeue) {
goto gt_CommandlineParameterError;
}
}
else
{
goto gt_CommandlineParameterError;
}
return true;
gt_CommandlineParameterError:
std::cerr << getTimeslipNow() << "コマンドライン引数の「--enqueue 送信先キュー名 寿命」「--dequeue 受信元キュー名 寿命」を漏れなく指定してください。" << std::endl
<< "例: --enqueue 1117 durable autodelete --dequeue 1116 durable autodelete" << std::endl
<< "寿命は可変個数設定可能「durable」「autodelete」「passive」「exclusive」" << std::endl
<< "name_enqueue=[" << name_enqueue << "]" << std::endl
<< "name_dequeue=[" << name_dequeue << "]" << std::endl;
return false;
}
}
#else
// バックグラウンド・プロセスに切り替えて使用する可能性も考慮した入出力API
namespace service_io
{
static int cleanMessage() {
service_io::oss.str(std::string());// 空文字列を入れてクリアーする
return 0;
}
static std::string dequeue() {
std::string cmd;
if (!std::getline(std::cin, cmd))
{
cmd = "quit";
}
return cmd;
}
}
#endif
class ServiceIo
{
public:
ServiceIo();
~ServiceIo();
};
#endif // #ifndef APERY_SERVICE_IO_HPP
service_io.cpp
//#include "stdafx.h"
#include "service_io.hpp"
ServiceIo::ServiceIo()
{
}
ServiceIo::~ServiceIo()
{
}
C++言語を使い慣れない間に 改造した。