/*
* GenericShell.cc
*
* Generic interactive shell
* Copyright (c) 2008, 2013, 2014 Linas Vepstas <linas@linas.org>
*
* This program is free software; you can redistribute it and/or modify
* it under the terms of the GNU Affero General Public License v3 as
* published by the Free Software Foundation and including the exceptions
* at http://opencog.org/wiki/Licenses
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU General Public License for more details.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program; if not, write to:
* Free Software Foundation, Inc.,
* 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
*/
#include <sys/prctl.h>
#include <chrono>
#include <mutex>
#include <thread>
#include <opencog/util/Logger.h>
#include <opencog/util/oc_assert.h>
#include <opencog/network/ConsoleSocket.h>
#include <opencog/network/SocketManager.h>
#include <opencog/eval/GenericEval.h>
#include "GenericShell.h"
using namespace opencog;
// Some random RFC 854 characters
#define IAC 0xff // Telnet Interpret As Command
#define TEOF 0xec // Telnet EOF
#define SUSP 0xed // Telnet suspend
#define ABRT 0xee // Telnet abort
#define NOP 0xf1 // Telnet NOP no-op
#define BRK 0xf3 // Telnet break
#define IP 0xf4 // Telnet IP Interrupt Process
#define AO 0xf5 // Telnet AO Abort Output
#define AYT 0xf6 // Telnet AYT Are You There
#define EC 0xf7 // Telnet EC Erase Character
#define EL 0xf8 // Telnet EL Erase Line
#define GA 0xf9 // Telnet GA Go ahead
#define WILL 0xfb // Telnet WILL
#define WONT 0xfc // Telnet WONT
#define DO 0xfd // Telnet DO
#define DONT 0xfe // Telnet DONT
#define TRANSMIT_BINARY 0 // Telnet RFC 856 8-bit-clean binary transmission
#define RFC_ECHO 1 // Telnet RFC 857 ECHO option
#define SUPPRESS_GO_AHEAD 3 // Telnet RFC 858 supporess go ahead
#define TIMING_MARK 6 // Telnet RFC 860 timing mark
#define LINEMODE 34 // Telnet RFC 1116 linemode
#define CHARSET 44 // Telnet RFC 2066
// Some random ASCII control characters (unix semantics)
#define EOT 0x4 // end or ^D at keyboard.
#define SYN 0x16 // quit or ^C at keyboard.
#define CAN 0x18 // cancel or ^X at keyboard.
#define ESC 0x1b // ecsape or ^[ at keyboard.
GenericShell::GenericShell(void) :
socket(nullptr),
evalthr(nullptr),
pollthr(nullptr),
_init_done(false),
abort_prompt("> "),
normal_prompt(abort_prompt),
pending_prompt("... "),
show_output(true),
show_prompt(true),
self_destruct(false),
apply_discipline(true),
_eval_done(true),
_evaluator(nullptr),
_name("gnrc")
{}
GenericShell::~GenericShell()
{
self_destruct = true;
// Wake up eval thread without cancelling the queue.
// We cannot use cancel() here because it prevents any further
// push() operations. This creates a race: if the socket is still
// draining buffers and trying to queue commands, those pushes
// will fail. Instead, push an empty string to wake up the thread.
try { evalque.push(""); }
catch (const std::exception& ex) {}
if (evalthr)
{
logger().debug("[GenericShell] dtor, wait for eval thread 0x%x.",
evalthr->native_handle());
evalthr->join();
logger().debug("[GenericShell] dtor, joined eval thread");
delete evalthr;
evalthr = nullptr;
}
logger().debug("[GenericShell] dtor finished.");
}
/* ============================================================== */
void GenericShell::hush_output(bool hush)
{
show_output = !hush;
}
void GenericShell::hush_prompt(bool hush)
{
show_prompt = !hush;
}
void GenericShell::discipline(bool d)
{
apply_discipline = d;
}
const std::string& GenericShell::get_prompt(void)
{
static const std::string empty_prompt = "";
if (!show_prompt) return empty_prompt;
// Use different prompts, depending on whether there is pending
// input or not.
if (_evaluator and _evaluator->input_pending())
{
return pending_prompt;
}
else
{
return normal_prompt;
}
}
/* ============================================================== */
/**
* Register this shell with the console.
*/
void GenericShell::set_socket(ConsoleSocket *s)
{
OC_ASSERT(socket==nullptr, "Shell already associated with socket!");
socket = s;
socket->SetShell(this);
}
/* ============================================================== */
static std::mutex _stdout_redirect_mutex;
#ifdef PERFORM_STDOUT_DUPLICATION
static GenericShell* _redirector = nullptr;
#endif
// Implementation requirements:
//
// 1) We want all evaluations to be carried out in serial order,
// so that the previous expression is fully evaluated before the
// next one is started.
// 2) We want all evaluations to be interruptible, so that if an
// expression is an infinite loop (or simply is taking too long)
// the user can send a control-C and interrupt the execution.
// 3) Due to the client-server socket model, this method should
// return as soon as possible, so that the caller can resume
// waiting on the socket, in case the user is trying to send
// a control-C to us.
// 4) Long-running evaluations should send output back to the user
// synchronously: i.e. send output back to the socket as the
// output is generated, instead of waiting for the evaluation
// to terminate first, before relaying output.
//
// The above requirements force us to create not just one, but two
// threads for each evaluation: one thread for the evaluation, and
// another thread to listen for results, and pass them on.
//
// Side-note: the constructor for this class runs in a different thread
// than the caller for this method. That's because the socket listen
// and socket accept runs in a different thread, than the socket
// receive. The receiver thread calls us.
//
void GenericShell::eval(const std::string &expr)
{
assert (not self_destruct);
// First time through, initialize the evaluator. We can't do this
// in the ctor, since we can't get the evaluator until after the
// derived-class ctor has run, and thus informed us as to whether
// the evaluator will be guile (scheme) or cython (python).
if (nullptr == _evaluator)
{
_init_done = false;
// Run the evaluation loop in a distinct thread.
auto eval_wrapper = [&](void) { eval_loop(); };
evalthr = new std::thread(eval_wrapper);
while (not _init_done) { sched_yield(); }
}
// Queue up the expr, where it will be evaluated in another thread.
if (apply_discipline)
line_discipline(expr);
else
enqueue_work(expr);
// The user is exiting the shell. No one will ever call a method on
// this instance ever again. So stop hogging space, and self-destruct.
// We have to do this here; there is no other opportunity to call dtor.
if (self_destruct)
{
// The atomspace-cog unit tests are unhappy if we don't send
// some reply message; they perceive an abnormal socket close.
// So we send them some garbage. They're using the hush sexpr
// prompt, so sending a lonely newline will get filtered.
socket->Send("goodbye!\n");
delete this;
}
}
/**
* Handle user-generated interrupt (ctrl-C, etc)
*/
void GenericShell::user_interrupt()
{
// Discard all pending, unevaluated junk in the queue.
// Failure to do so will typically result in confusing
// the shell user.
std::string junk;
while (not evalque.is_empty()) evalque.try_pop(junk);
// Work around timing window, where queue was just now emptied,
// but the scheme evaluator has not yet started... and so the
// command that is to be interrupted hasn't even to begun to
// execute, when we go to interrupt it.
usleep(10000);
// It can also happen that eval_loop() has exited, and has set
// _evaluator to nullptr. So we need to check for this. Of course,
// this is insane and racey, and in principle, we need some kind
// additional condition variable to avoid hitting this during shutdown.
// but for now, just punt.
//
// This happens if you run TopShell, enter . carriage return ctrl-C
// and then boom, here we are.
if (nullptr == _evaluator)
{
logger().info("[GenericShell] user interrupt after shutdown!");
return;
}
_evaluator->interrupt();
_evaluator->clear_pending();
put_output(abort_prompt);
finish_eval();
}
/* ============================================================== */
/**
* Handle special characters, evaluate the expression.
*/
void GenericShell::line_discipline(const std::string &expr)
{
size_t len = expr.length();
logger().debug("[GenericShell] line disc: expr, len of %zd ='%s'",
len, expr.c_str());
if (0 == len)
{
enqueue_work("\n");
return;
}
// Handle Telnet RFC 854 IAC format
// Basically, we're looking for telnet-encoded abort or interrupt
// characters, starting at the end of the input string. If they
// are there, then don't process input, and clear out the evaluator.
// Also, be sure to send telnet IAC WILL TIMING-MARK so that telnet
// doesn't sit there discarding output forever.
//
// Search for IAC to at most 20 chars from the end of the string.
int i = len-2;
int m = len - 20;
if (m < 0) m = 0;
while (m <= i)
{
unsigned char c = expr[m];
if (IAC == c) break;
m++;
}
unsigned char breakpt = 0;
bool got_break = false;
while (m <= i)
{
unsigned char c = expr[m];
if (IAC != c)
{
logger().debug("[GenericShell] Garbled telnet cmd=%d at pos=%d of %s",
c, m, expr.c_str());
put_output(get_prompt());
return;
}
c = expr[m+1];
if (IP == c or AO == c or SUSP == c)
{
logger().debug("[GenericShell] got telnet IAC user-interrupt %d", c);
// Must send TIMING-MARK first, as otherwise telnet
// silently ignores (i.e. does not send to us) any bytes
// that come after it, per RFC 860.
unsigned char ok[] = {IAC, WILL, TIMING_MARK, '\n', 0};
put_output((const char *) ok);
user_interrupt();
return;
}
// Erase line -- just ignore this line.
// Also other things we want to ignore, like abort.
if (EL == c or EC == c or ABRT == c or
AYT == c or GA == c or NOP == c)
{
logger().debug("[GenericShell] Ignoring telnet IAC %d", c);
put_output(get_prompt());
return;
}
// End-of-file just like ctrl-D
if (TEOF == c)
{
logger().debug("[GenericShell] Got end-of-file; exiting shell");
self_destruct = true;
evalque.cancel();
if (show_prompt)
put_output("Exiting the shell\n");
return;
}
// Break
if (BRK == c)
{
logger().debug("[GenericShell] Received IAC BREAK");
got_break = true;
breakpt = m;
m += 2;
continue;
}
if (DO == c)
{
unsigned char a = expr[m+2];
// Some telnets, including on Debian Stable, send us
// IAC DO TIMING-MARK instead of a IAC IP or IAC AO
// when the user hits ctrl-C. This seems broken to me,
// but whatever. Pretend its a normal interrupt.
if (TIMING_MARK == a)
{
logger().debug("[GenericShell] Received IAC timing-mark");
// Send TIMING-MARK first, as otherwise telnet silently
// ignores any bytes that come before it, per RFC 860.
unsigned char ok[] = {IAC, WILL, TIMING_MARK, '\n', 0};
put_output((const char *) ok);
// Crap. For some reason, telnet will garble some but not
// all UTF-8. For example: (Concept "“") which is e2 80 9c
// but tlenet sends e2 80 ff f3 ff fd 6 or, that is,
// IAC BRK IAC DO TIMING_MARK. And never sends the rest.
// See for example:
// https://www.reddit.com/r/node/comments/1op7ns/bizarre_issues_with_utf8_over_telnet/
// I cannot find a solution. WILL/WONT/DO/DONT above makes
// no difference.
if (got_break)
{
std::string mute = expr;
mute[breakpt] = 0;
logger().warn("[GenericShell] Telnet sent RFC 860 TIMING MARK -- Probably garbled UTF-8: %s",
mute.c_str());
enqueue_work(mute);
return;
}
logger().debug("[GenericShell] Huh? Assume user-interrupt");
user_interrupt();
return;
}
// If telnet ever tries to go into character mode,
// it will send us SUPPRESS-GO-AHEAD and ECHO. Try to
// stop that, we don't want to effing fiddle with that.
if (SUPPRESS_GO_AHEAD == a)
{
unsigned char ok[] = {IAC, WILL, SUPPRESS_GO_AHEAD, 0};
put_output((const char *) ok);
return;
}
if (RFC_ECHO == a)
{
unsigned char ok[] = {IAC, WONT, RFC_ECHO, '\n', 0};
put_output((const char *) ok);
return;
}
logger().debug("[GenericShell] IAC WONT %d", a);
unsigned char ok[] = {IAC, WONT, a, '\n', 0};
put_output((const char *) ok);
return;
}
if (WILL == c)
{
// Refuse to perform sub-negotation when
// IAC WILL LINEMODE is sent by the telnet client.
unsigned char a = expr[m+2];
if (LINEMODE == a)
{
unsigned char ok[] = {IAC, DONT, LINEMODE, '\n', 0};
put_output((const char *) ok);
return;
}
// Ignore anything else.
logger().debug("[GenericShell] ignoring telnet IAC WILL %d", a);
return;
}
m += 2;
}
// Don't evaluate if the line is terminated by
// escape (^[), cancel (^X) or quit (^C)
// These would typically be sent by netcat, and not telnet.
unsigned char c = expr[len-1];
if ((SYN == c) || (CAN == c) || (ESC == c))
{
logger().debug("[GenericShell] got user-interrupt %d", c);
user_interrupt();
return;
}
// Look for either an isolated control-D, or a single period on a line
// by itself. This means "leave the shell". We leave the shell by
// unsetting the shell pointer in the ConsoleSocket.
// 0x4 is ASCII EOT, which is what ctrl-D at keybd becomes.
if ((false == _evaluator->input_pending()) and
((EOT == expr[len-1]) or ((1 == len) and ('.' == expr[0]))))
{
logger().debug("[GenericShell] got control-D; exiting shell");
self_destruct = true;
evalque.cancel();
if (show_prompt)
put_output("Exiting the shell\n");
return;
}
/*
* The newline was cut by the request subsystem. Re-insert it;
* otherwise, comments within procedures will have the effect of
* commenting out the rest of the procedure, leading to garbage.
*
* XXX Is this still true?
*/
enqueue_work(expr + "\n");
}
/* ============================================================== */
// The problem being addressed here is that the shell destructor
// can start running (because the socket was closed) before the
// evaluator has even started running. Need to hold off.
void GenericShell::start_eval()
{
OC_ASSERT(_eval_done, "Bad evaluator flag state!");
std::unique_lock<std::mutex> lck(_eval_mtx);
_eval_done = false;
}
void GenericShell::finish_eval()
{
// Repeated control-C will send us here with _eval_done already set..
std::unique_lock<std::mutex> lck(_eval_mtx);
_eval_done = true;
_eval_cv.notify_all();
}
void GenericShell::while_not_done()
{
std::unique_lock<std::mutex> lck(_eval_mtx);
while (not _eval_done) _eval_cv.wait(lck);
}
/* ============================================================== */
/// eval_loop. Dequeue and run each queued evaluation request.
/// Assumes line discipline has already been done (as it must be:
/// it is impossible to queue OOB interrupts.)
void GenericShell::eval_loop(void)
{
prctl(PR_SET_NAME, "cogserv:eval", 0, 0, 0);
logger().debug("[GenericShell] enter eval loop");
OC_ASSERT(nullptr == _evaluator, "Bad evaluator state!");
// Per-shell evaluator. We do this here, not in the ctor, because
// we want a unique, private evaluator for this thread. By contrast,
// the ctor might be called many times within one thread, and thus,
// each invocation would end up with the same evaluator.
_evaluator = get_evaluator();
_evaluator->clear_pending();
// Poll for output from the evaluator, and send back results.
auto poll_wrapper = [&](void) { poll_loop(); };
pollthr = new std::thread(poll_wrapper);
// Derived-class initializer. (None of the shells use this
// at this time ... this is left over from earlier times.)
thread_init();
// Go through the body of the loop at least once.
std::string in;
do
{
try
{
// Do not begin the next queued expr until the last
// has finished. Failure to do this can result in
// weird crashes in the SchemeEval class.
while_not_done();
wake_poll();
// Note that this pop will stall until the queue
// becomes non-empty.
evalque.pop(in);
// Skip empty strings - these are sentinel wake-up signals.
if (in.empty())
continue;
logger().debug("[GenericShell] start eval %s of '%s'",
_evaluator->get_name().c_str(), in.c_str());
wake_poll();
start_eval();
_evaluator->begin_eval();
_evaluator->eval_expr(in);
wake_poll();
}
catch (const RuntimeException& ex)
{
/* Python throws these on user syntax errors.*/
/* Python sometimes deadlocks. Don't know why. */
_eval_done = true;
_poll_mtx.unlock();
}
catch (const concurrent_queue<std::string>::Canceled& ex)
{
break;
}
} while (not self_destruct);
// If we are here, then we can safely assume that the socket has
// been closed, that the dtor for this instance has been called.
// However, there may still be some remaining, unfinished work
// in the command queue; drain the queue, before shutting down.
assert(self_destruct);
evalque.cancel_reset();
// Let the polling thread die first. If we don't do this, it will
// interfere with the manual polling below.
pollthr->join();
delete pollthr;
pollthr = nullptr;
// Nothing more will be queued, so we can safely loop over remainder
// of the queue, without any additional need for locking/waiting.
while (0 < evalque.size())
{
// As mentioned before, do not begin the next queued expr until
// the last has finished. Failure to do this results in crashes.
poll_and_send();
while (not _eval_done)
{
usleep(10000);
poll_and_send();
}
try
{
evalque.pop(in);
}
catch (const concurrent_queue<std::string>::Canceled& ex)
{
evalque.cancel_reset();
continue;
}
// Skip empty strings - these are sentinel wake-up signals.
if (in.empty())
continue;
logger().debug("[GenericShell] finishing; eval of '%s'", in.c_str());
start_eval();
_evaluator->begin_eval();
_evaluator->eval_expr(in);
}
// Continue polling until the evaluation really is done.
poll_and_send();
while (not _eval_done)
{
usleep(10000);
poll_and_send();
}
socket->SetShell(nullptr);
// After we exit, the _evaluator will be reclaimed by the
// thread dtor running in the evaluator pool.
_evaluator = nullptr;
logger().debug("[GenericShell] exit eval loop");
}
void GenericShell::poll_and_send(void)
{
std::string retstr(poll_output());
if (0 < retstr.size())
socket->Send(retstr);
}
void GenericShell::wake_poll(void)
{
std::unique_lock<std::mutex> lck(_poll_mtx);
_poll_cv.notify_all();
}
void GenericShell::poll_loop(void)
{
_init_done = true;
prctl(PR_SET_NAME, "cogserv:poll", 0, 0, 0);
std::unique_lock<std::mutex> lock(_poll_mtx);
// Poll for output from the evaluator, and send back results.
while (not self_destruct)
{
// That's right, call this twice in a row.
// This avoids a stall in the _poll_cv below.
poll_and_send();
poll_and_send();
// Continue polling, about 100 times per second, even if
// evaluation of the the previous expr is completed. It
// might have started some long-running thread that is
// continuing to print, and we want to forward those
// prints to the user. (Its pointless to poll faster or
// slower than this... except ...)
// ... except ...
// Just after finishing, there might be another command in
// the queue, and so we want to respond to that, as soon as
// it seems to be done.
//
using namespace std::chrono_literals;
_poll_cv.wait_for(lock, 10ms);
}
lock.unlock();
// It's also possible that another thread reaches the dtor
// (setting self_destruct == true) shortly after an evaluation
// has just finished. It may then exit the loop above without
// polling the output, eventually causing the evalthr to stay
// in while_not_done() forever. So let's poll again, one more
// time, here.
usleep(250);
do
{
poll_and_send();
usleep(10000);
}
while (not _eval_done);
}
void GenericShell::thread_init(void)
{
/* No-op. The Scheme shell sets the current atomspace here */
}
/* ============================================================== */
void GenericShell::put_output(const std::string& s)
{
std::lock_guard<std::mutex> lock(_pending_mtx);
_pending_output += s;
}
std::string GenericShell::get_output()
{
std::lock_guard<std::mutex> lock(_pending_mtx);
std::string result = _pending_output;
_pending_output.clear();
return result;
}
std::string GenericShell::poll_output()
{
// If there's pending output, return that.
std::string pend(get_output());
if (0 < pend.size()) return pend;
// If we are here, there's no pending output. Does the evaluator
// have anything for us? Note that the ->poll_result() method
// will block, if the evaluator is not done. Note that we must
// do the get_output() again, else ctrl-C's will not be returned
// in proper order to a telnet connection.
std::string result(_evaluator->poll_result());
if (0 < result.size())
return get_output() + result;
// If we are here, the evaluator is done. Return shell prompts.
if (_eval_done)
{
// Record evaluator errors. Automated scripts pumping the cogserver
// with data might not always notice these, so at least record them
// in the log file, where they might get noticed by some human.
if (_evaluator->eval_error())
{
std::string errmsg(_evaluator->get_error_string());
_evaluator->clear_pending(); // clear the error bit.
logger().info("[GenericShell] evaluator error:\n%s", errmsg.c_str());
if (show_output) return errmsg;
}
return "";
}
// If we are here, the result size was zero, and so we know we are
// done, but the _eval_done flag had not been set. Use this to wake
// up everyone.
finish_eval();
if (show_output and show_prompt)
{
if (_evaluator->input_pending())
return pending_prompt;
return normal_prompt;
}
return "";
}
void GenericShell::enqueue_work(const std::string& expr)
{
// New work cannot be enqueued as long as there is a barrier in place.
socket->get_socket_manager()->block_on_bar();
// If the cogserver is shutting down, while there are still
// open sockets with data on them, we might end up here, trying
// to queue up work onto a closed queue. This is a race, and
// curing that race in some other way is ... not worth the effort.
// The goal here is to not crash with an uncaught exception.
try { evalque.push(expr); }
catch (const concurrent_queue<std::string>::Canceled& ex) {}
}
/* ===================== END OF FILE ============================ */