session.cc 17 KB
Newer Older
1 2 3 4 5 6 7 8 9 10 11 12 13 14
// Copyright (C) 2009  Internet Systems Consortium, Inc. ("ISC")
//
// Permission to use, copy, modify, and/or distribute this software for any
// purpose with or without fee is hereby granted, provided that the above
// copyright notice and this permission notice appear in all copies.
//
// THE SOFTWARE IS PROVIDED "AS IS" AND ISC DISCLAIMS ALL WARRANTIES WITH
// REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY
// AND FITNESS.  IN NO EVENT SHALL ISC BE LIABLE FOR ANY SPECIAL, DIRECT,
// INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM
// LOSS OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE
// OR OTHER TORTIOUS ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR
// PERFORMANCE OF THIS SOFTWARE.

15
#include <config.h>
JINMEI Tatuya's avatar
JINMEI Tatuya committed
16
#include <cc/session_config.h>
17
#include <cc/logger.h>
18

19
#include <stdint.h>
20

21 22 23 24 25 26
// XXX: there seems to be a strange dependency between ASIO and std library
// definitions.  On some platforms if we include std headers before ASIO
// headers unexpected behaviors will happen.
// A middle term solution is to generalize our local wrapper interface
// (currently only available for the auth server), where all such portability
// issues are hidden, and to have other modules use the wrapper.
27
#include <unistd.h>             // for some IPC/network system calls
28 29
#include <asio.hpp>
#include <asio/error_code.hpp>
30
#include <asio/deadline_timer.hpp>
31 32
#include <asio/system_error.hpp>

33
#include <cstdio>
34
#include <vector>
35 36 37
#include <iostream>
#include <sstream>

JINMEI Tatuya's avatar
JINMEI Tatuya committed
38 39
#include <sys/un.h>

40
#include <boost/bind.hpp>
41
#include <boost/optional.hpp>
42
#include <boost/function.hpp>
43
#include <boost/date_time/posix_time/posix_time_types.hpp>
44

45 46
#include <exceptions/exceptions.h>

JINMEI Tatuya's avatar
JINMEI Tatuya committed
47 48
#include <cc/data.h>
#include <cc/session.h>
49

50
using namespace std;
Jelte Jansen's avatar
Jelte Jansen committed
51 52
using namespace isc::cc;
using namespace isc::data;
53

54
// some of the asio names conflict with socket API system calls
Jelte Jansen's avatar
 
Jelte Jansen committed
55
// (e.g. write(2)) so we don't import the entire asio namespace.
56
using asio::io_service;
57

Jelte Jansen's avatar
Jelte Jansen committed
58
namespace {
59 60 61 62 63
/// \brief Sets the given Optional 'result' to the given error code
/// Used as a callback for emulating sync reads with async calls
/// \param result Pointer to the optional to set
/// \param err The error code to set it to
void
JINMEI Tatuya's avatar
JINMEI Tatuya committed
64 65 66
setResult(boost::optional<asio::error_code>* result,
          const asio::error_code& err)
{
67 68
    result->reset(err);
}
Jelte Jansen's avatar
Jelte Jansen committed
69 70
}

71 72
namespace isc {
namespace cc {
73

74 75
class SessionImpl {
public:
JINMEI Tatuya's avatar
JINMEI Tatuya committed
76 77
    SessionImpl(io_service& io_service) :
        sequence_(-1), queue_(Element::createList()),
78
        io_service_(io_service), socket_(io_service_), data_length_(0),
Jelte Jansen's avatar
Jelte Jansen committed
79
        timeout_(MSGQ_DEFAULT_TIMEOUT)
JINMEI Tatuya's avatar
JINMEI Tatuya committed
80 81 82 83 84
    {}
    void establish(const char& socket_file);
    void disconnect();
    void writeData(const void* data, size_t datalen);
    size_t readDataLength();
85 86
    // Blocking read. Will throw a SessionTimeout if the timeout value
    // (in seconds) is thrown. If timeout is 0 it will block forever
JINMEI Tatuya's avatar
JINMEI Tatuya committed
87 88
    void readData(void* data, size_t datalen);
    void startRead(boost::function<void()> user_handler);
89 90
    void setTimeout(size_t seconds) { timeout_ = seconds; };
    size_t getTimeout() const { return timeout_; };
JINMEI Tatuya's avatar
JINMEI Tatuya committed
91

Jelte Jansen's avatar
Jelte Jansen committed
92
    long int sequence_; // the next sequence number to use
93
    std::string lname_;
Jelte Jansen's avatar
Jelte Jansen committed
94
    ElementPtr queue_;
95

96
private:
97
    void internalRead(const asio::error_code& error,
98 99 100 101
                      size_t bytes_transferred);

private:
    io_service& io_service_;
Jelte Jansen's avatar
 
Jelte Jansen committed
102
    asio::local::stream_protocol::socket socket_;
103 104
    uint32_t data_length_;
    boost::function<void()> user_handler_;
105
    asio::error_code error_;
106
    size_t timeout_;
Jelte Jansen's avatar
Jelte Jansen committed
107 108 109 110 111 112 113 114 115 116

    // By default, unless changed or disabled, blocking reads on
    // the msgq channel will time out after 4 seconds in this
    // implementation.
    // This number is chosen to be low enough so that whatever
    // component is blocking does not seem to be hanging, but
    // still gives enough time for other modules to respond if they
    // are busy. If this choice turns out to be a bad one, we can
    // change it later.
    static const size_t MSGQ_DEFAULT_TIMEOUT = 4000;
117
};
118

119
void
JINMEI Tatuya's avatar
JINMEI Tatuya committed
120
SessionImpl::establish(const char& socket_file) {
121
    try {
122
        LOG_DEBUG(logger, DBG_TRACE_BASIC, CC_ESTABLISH).arg(&socket_file);
JINMEI Tatuya's avatar
JINMEI Tatuya committed
123 124
        socket_.connect(asio::local::stream_protocol::endpoint(&socket_file),
                        error_);
125
        LOG_DEBUG(logger, DBG_TRACE_BASIC, CC_ESTABLISHED);
JINMEI Tatuya's avatar
JINMEI Tatuya committed
126
    } catch(const asio::system_error& se) {
127
        LOG_FATAL(logger, CC_CONN_ERROR).arg(se.what());
128 129
        isc_throw(SessionError, se.what());
    }
130
    if (error_) {
131
        LOG_FATAL(logger, CC_NO_MSGQ).arg(error_.message());
JINMEI Tatuya's avatar
JINMEI Tatuya committed
132 133
        isc_throw(SessionError, "Unable to connect to message queue: " <<
                  error_.message());
134
    }
135 136 137
}

void
JINMEI Tatuya's avatar
JINMEI Tatuya committed
138
SessionImpl::disconnect() {
139
    LOG_DEBUG(logger, DBG_TRACE_BASIC, CC_DISCONNECT);
140 141
    socket_.close();
    data_length_ = 0;
142 143 144
}

void
JINMEI Tatuya's avatar
JINMEI Tatuya committed
145
SessionImpl::writeData(const void* data, size_t datalen) {
146
    try {
147
        asio::write(socket_, asio::buffer(data, datalen));
148
    } catch (const asio::system_error& asio_ex) {
149
        LOG_FATAL(logger, CC_WRITE_ERROR).arg(asio_ex.what());
150
        isc_throw(SessionError, "ASIO write failed: " << asio_ex.what());
151
    }
152 153
}

154
size_t
JINMEI Tatuya's avatar
JINMEI Tatuya committed
155
SessionImpl::readDataLength() {
156 157 158 159 160
    size_t ret_len = data_length_;
    
    if (ret_len == 0) {
        readData(&data_length_, sizeof(data_length_));
        if (data_length_ == 0) {
161
            LOG_ERROR(logger, CC_LENGTH_NOT_READY);
162 163 164 165 166 167 168 169 170 171
            isc_throw(SessionError, "ASIO read: data length is not ready");
        }
        ret_len = ntohl(data_length_);
    }

    data_length_ = 0;
    return (ret_len);
}

void
JINMEI Tatuya's avatar
JINMEI Tatuya committed
172
SessionImpl::readData(void* data, size_t datalen) {
173 174 175
    boost::optional<asio::error_code> read_result;
    boost::optional<asio::error_code> timer_result;

176
    try {
177
        asio::async_read(socket_, asio::buffer(data, datalen),
178
                         boost::bind(&setResult, &read_result, _1));
179
        asio::deadline_timer timer(socket_.io_service());
180

181 182
        if (getTimeout() != 0) {
            timer.expires_from_now(boost::posix_time::milliseconds(getTimeout()));
183
            timer.async_wait(boost::bind(&setResult, &timer_result, _1));
184
        }
185

186 187 188 189
        // wait until either we have read the data we want, the
        // timer expires, or one of the two is triggered with an error.
        // When one of them has a result, cancel the other, and wait
        // until the cancel is processed before we continue
190
        while (!read_result && !timer_result) {
191
            socket_.io_service().run_one();
Jelte Jansen's avatar
Jelte Jansen committed
192 193

            // Don't cancel the timer if we haven't set it
194
            if (read_result && getTimeout() != 0) {
195
                timer.cancel();
196 197 198
                while (!timer_result) {
                    socket_.io_service().run_one();
                }
199 200
            } else if (timer_result) {
                socket_.cancel();
201 202 203
                while (!read_result) {
                    socket_.io_service().run_one();
                }
204 205
            }
        }
206

207 208
        // asio::error_code evaluates to false if there was no error
        if (*read_result) {
209
            if (*read_result == asio::error::operation_aborted) {
210
                LOG_ERROR(logger, CC_TIMEOUT);
211 212 213
                isc_throw(SessionTimeout,
                          "Timeout while reading data from cc session");
            } else {
214
                LOG_ERROR(logger, CC_READ_ERROR).arg(read_result->message());
215 216 217 218
                isc_throw(SessionError,
                          "Error while reading data from cc session: " <<
                          read_result->message());
            }
219
        }
220
    } catch (const asio::system_error& asio_ex) {
221
        // to hide ASIO specific exceptions, we catch them explicitly
222
        // and convert it to SessionError.
223
        LOG_FATAL(logger, CC_READ_EXCEPTION).arg(asio_ex.what());
224
        isc_throw(SessionError, "ASIO read failed: " << asio_ex.what());
225 226 227 228
    }
}

void
JINMEI Tatuya's avatar
JINMEI Tatuya committed
229
SessionImpl::startRead(boost::function<void()> user_handler) {
230 231
    data_length_ = 0;
    user_handler_ = user_handler;
232 233 234 235 236
    asio::async_read(socket_, asio::buffer(&data_length_,
                                           sizeof(data_length_)),
                     boost::bind(&SessionImpl::internalRead, this,
                                 asio::placeholders::error,
                                 asio::placeholders::bytes_transferred));
237 238 239
}

void
JINMEI Tatuya's avatar
JINMEI Tatuya committed
240
SessionImpl::internalRead(const asio::error_code& error,
241
                          size_t bytes_transferred)
242
{
243 244 245 246
    if (!error) {
        assert(bytes_transferred == sizeof(data_length_));
        data_length_ = ntohl(data_length_);
        if (data_length_ == 0) {
247
            LOG_ERROR(logger, CC_ZERO_LENGTH);
248 249 250 251
            isc_throw(SessionError, "Invalid message length (0)");
        }
        user_handler_();
    } else {
252
        LOG_ERROR(logger, CC_ASYNC_READ_FAILED).arg(error.value());
253 254 255 256
        isc_throw(SessionError, "asynchronous read failed");
    }
}

257 258
Session::Session(asio::io_service& io_service) :
    impl_(new SessionImpl(io_service))
JINMEI Tatuya's avatar
JINMEI Tatuya committed
259 260 261 262 263 264 265 266 267 268 269 270 271
{}

Session::~Session() {
    delete impl_;
}

void
Session::disconnect() {
    impl_->disconnect();
}

void
Session::startRead(boost::function<void()> read_callback) {
272
    LOG_DEBUG(logger, DBG_TRACE_DETAILED, CC_START_READ);
JINMEI Tatuya's avatar
JINMEI Tatuya committed
273 274
    impl_->startRead(read_callback);
}
275 276

namespace {                     // maybe unnecessary.
277 278
// This is a helper class to make the establish() method (below) exception-safe
// with the RAII approach.
279
class SessionHolder {
280
public:
281 282 283 284 285 286 287 288 289
    SessionHolder(SessionImpl* obj) : impl_obj_(obj) {}
    ~SessionHolder()
    {
        if (impl_obj_ != NULL) {
            impl_obj_->disconnect();
        }
    }
    void clear() { impl_obj_ = NULL; }
    SessionImpl* impl_obj_;
290
};
291 292
}

293
void
294
Session::establish(const char* socket_file) {
295 296 297 298 299 300 301 302
    if (socket_file == NULL) {
        socket_file = getenv("BIND10_MSGQ_SOCKET_FILE");
    }
    if (socket_file == NULL) {
        socket_file = BIND10_MSGQ_SOCKET_FILE;
    }

    impl_->establish(*socket_file);
303 304 305 306 307

    // once established, encapsulate the implementation object so that we
    // can safely release the internal resource when exception happens
    // below.
    SessionHolder session_holder(impl_);
308 309 310 311

    //
    // send a request for our local name, and wait for a response
    //
312
    ConstElementPtr get_lname_msg =
313
        Element::fromJSON("{ \"type\": \"getlname\" }");
314 315
    sendmsg(get_lname_msg);

316
    ConstElementPtr routing, msg;
317
    recvmsg(routing, msg, false);
318

319 320
    impl_->lname_ = msg->get("lname")->stringValue();

321 322
    // At this point there's no risk of resource leak.
    session_holder.clear();
323 324 325
}

//
JINMEI Tatuya's avatar
JINMEI Tatuya committed
326 327
// Convert to wire format and send this via the stream socket with its length
// prefix.
328 329
//
void
330
Session::sendmsg(ConstElementPtr msg) {
331
    std::string header_wire = msg->toWire();
332
    unsigned int length = 2 + header_wire.length();
333
    unsigned int length_net = htonl(length);
334 335
    unsigned short header_length = header_wire.length();
    unsigned short header_length_net = htons(header_length);
336

337 338 339
    impl_->writeData(&length_net, sizeof(length_net));
    impl_->writeData(&header_length_net, sizeof(header_length_net));
    impl_->writeData(header_wire.data(), header_length);
340 341
}

342
void
343
Session::sendmsg(ConstElementPtr env, ConstElementPtr msg) {
344 345
    std::string header_wire = env->toWire();
    std::string body_wire = msg->toWire();
346 347 348 349 350
    unsigned int length = 2 + header_wire.length() + body_wire.length();
    unsigned int length_net = htonl(length);
    unsigned short header_length = header_wire.length();
    unsigned short header_length_net = htons(header_length);

351 352 353 354
    impl_->writeData(&length_net, sizeof(length_net));
    impl_->writeData(&header_length_net, sizeof(header_length_net));
    impl_->writeData(header_wire.data(), header_length);
    impl_->writeData(body_wire.data(), body_wire.length());
355 356
}

357
bool
358 359
Session::recvmsg(ConstElementPtr& msg, bool nonblock, int seq) {
    ConstElementPtr l_env;
JINMEI Tatuya's avatar
JINMEI Tatuya committed
360
    return (recvmsg(l_env, msg, nonblock, seq));
361 362
}

363
bool
364 365 366
Session::recvmsg(ConstElementPtr& env, ConstElementPtr& msg,
                 bool nonblock, int seq)
{
367
    size_t length = impl_->readDataLength();
Jelte Jansen's avatar
Jelte Jansen committed
368
    if (hasQueuedMsgs()) {
369
        ConstElementPtr q_el;
370
        for (size_t i = 0; i < impl_->queue_->size(); i++) {
Jelte Jansen's avatar
Jelte Jansen committed
371 372 373 374 375 376 377 378 379 380 381
            q_el = impl_->queue_->get(i);
            if (( seq == -1 &&
                  !q_el->get(0)->contains("reply")
                ) || (
                  q_el->get(0)->contains("reply") &&
                  q_el->get(0)->get("reply")->intValue() == seq
                )
               ) {
                   env = q_el->get(0);
                   msg = q_el->get(1);
                   impl_->queue_->remove(i);
JINMEI Tatuya's avatar
JINMEI Tatuya committed
382
                   return (true);
Jelte Jansen's avatar
Jelte Jansen committed
383 384 385 386
            }
        }
    }
    
387 388
    unsigned short header_length_net;
    impl_->readData(&header_length_net, sizeof(header_length_net));
389 390

    unsigned short header_length = ntohs(header_length_net);
391
    if (header_length > length || length < 2) {
392
        LOG_ERROR(logger, CC_INVALID_LENGTHS).arg(length).arg(header_length);
393 394 395
        isc_throw(SessionError, "Length parameters invalid: total=" << length
                  << ", header=" << header_length);
    }
396 397 398

    // remove the header-length bytes from the total length
    length -= 2;
399
    std::vector<char> buffer(length);
400
    impl_->readData(&buffer[0], length);
401

402
    std::string header_wire = std::string(&buffer[0], header_length);
403 404
    std::string body_wire = std::string(&buffer[0] + header_length,
                                        length - header_length);
405 406
    std::stringstream header_wire_stream;
    header_wire_stream << header_wire;
407 408
    ConstElementPtr l_env =
        Element::fromWire(header_wire_stream, header_length);
409
    
410 411
    std::stringstream body_wire_stream;
    body_wire_stream << body_wire;
412 413
    ConstElementPtr l_msg =
        Element::fromWire(body_wire_stream, length - header_length);
Jelte Jansen's avatar
Jelte Jansen committed
414 415 416 417 418 419 420 421 422
    if ((seq == -1 &&
         !l_env->contains("reply")
        ) || (
         l_env->contains("reply") &&
         l_env->get("reply")->intValue() == seq
        )
       ) {
        env = l_env;
        msg = l_msg;
JINMEI Tatuya's avatar
JINMEI Tatuya committed
423
        return (true);
Jelte Jansen's avatar
Jelte Jansen committed
424
    } else {
425
        ElementPtr q_el = Element::createList();
Jelte Jansen's avatar
Jelte Jansen committed
426 427 428
        q_el->add(l_env);
        q_el->add(l_msg);
        impl_->queue_->add(q_el);
JINMEI Tatuya's avatar
JINMEI Tatuya committed
429
        return (recvmsg(env, msg, nonblock, seq));
Jelte Jansen's avatar
Jelte Jansen committed
430
    }
431 432 433
    // XXXMLG handle non-block here, and return false for short reads
}

434
void
435
Session::subscribe(std::string group, std::string instance) {
436
    LOG_DEBUG(logger, DBG_TRACE_DETAILED, CC_SUBSCRIBE).arg(group);
437
    ElementPtr env = Element::createMap();
438 439 440 441 442 443 444 445 446

    env->set("type", Element::create("subscribe"));
    env->set("group", Element::create(group));
    env->set("instance", Element::create(instance));

    sendmsg(env);
}

void
447
Session::unsubscribe(std::string group, std::string instance) {
448
    LOG_DEBUG(logger, DBG_TRACE_DETAILED, CC_UNSUBSCRIBE).arg(group);
449
    ElementPtr env = Element::createMap();
450 451 452 453 454 455 456 457

    env->set("type", Element::create("unsubscribe"));
    env->set("group", Element::create(group));
    env->set("instance", Element::create(instance));

    sendmsg(env);
}

Jelte Jansen's avatar
Jelte Jansen committed
458
int
459
Session::group_sendmsg(ConstElementPtr msg, std::string group,
460
                       std::string instance, std::string to)
461
{
462 463
    LOG_DEBUG(logger, DBG_TRACE_DETAILED, CC_GROUP_SEND).arg(msg->str()).
        arg(group);
464
    ElementPtr env = Element::createMap();
Jelte Jansen's avatar
Jelte Jansen committed
465
    long int nseq = ++impl_->sequence_;
Jelte Jansen's avatar
Jelte Jansen committed
466
    
467
    env->set("type", Element::create("send"));
468
    env->set("from", Element::create(impl_->lname_));
469 470 471
    env->set("to", Element::create(to));
    env->set("group", Element::create(group));
    env->set("instance", Element::create(instance));
Jelte Jansen's avatar
Jelte Jansen committed
472
    env->set("seq", Element::create(nseq));
473
    //env->set("msg", Element::create(msg->toWire()));
474

475
    sendmsg(env, msg);
JINMEI Tatuya's avatar
JINMEI Tatuya committed
476
    return (nseq);
477 478 479
}

bool
480
Session::group_recvmsg(ConstElementPtr& envelope, ConstElementPtr& msg,
Jelte Jansen's avatar
Jelte Jansen committed
481
                       bool nonblock, int seq)
482
{
483 484 485 486 487 488 489 490 491
    LOG_DEBUG(logger, DBG_TRACE_DETAILED, CC_GROUP_RECEIVE);
    bool result(recvmsg(envelope, msg, nonblock, seq));
    if (result) {
        LOG_DEBUG(logger, DBG_TRACE_DETAILED, CC_GROUP_RECEIVED).
            arg(envelope->str()).arg(msg->str());
    } else {
        LOG_DEBUG(logger, DBG_TRACE_DETAILED, CC_NO_MESSAGE);
    }
    return (result);
492 493
}

Jelte Jansen's avatar
Jelte Jansen committed
494
int
495
Session::reply(ConstElementPtr envelope, ConstElementPtr newmsg) {
496 497
    LOG_DEBUG(logger, DBG_TRACE_DETAILED, CC_REPLY).arg(envelope->str()).
        arg(newmsg->str());
498
    ElementPtr env = Element::createMap();
Jelte Jansen's avatar
Jelte Jansen committed
499
    long int nseq = ++impl_->sequence_;
Jelte Jansen's avatar
Jelte Jansen committed
500
    
501
    env->set("type", Element::create("send"));
502
    env->set("from", Element::create(impl_->lname_));
503 504 505
    env->set("to", Element::create(envelope->get("from")->stringValue()));
    env->set("group", Element::create(envelope->get("group")->stringValue()));
    env->set("instance", Element::create(envelope->get("instance")->stringValue()));
Jelte Jansen's avatar
Jelte Jansen committed
506
    env->set("seq", Element::create(nseq));
507
    env->set("reply", Element::create(envelope->get("seq")->intValue()));
508

509
    sendmsg(env, newmsg);
510

JINMEI Tatuya's avatar
JINMEI Tatuya committed
511
    return (nseq);
512
}
Jelte Jansen's avatar
Jelte Jansen committed
513 514

bool
515
Session::hasQueuedMsgs() const {
Jelte Jansen's avatar
Jelte Jansen committed
516 517 518
    return (impl_->queue_->size() > 0);
}

519 520
void
Session::setTimeout(size_t milliseconds) {
521
    LOG_DEBUG(logger, DBG_TRACE_DETAILED, CC_SET_TIMEOUT).arg(milliseconds);
522 523 524 525
    impl_->setTimeout(milliseconds);
}

size_t
526
Session::getTimeout() const {
Jelte Jansen's avatar
Jelte Jansen committed
527
    return (impl_->getTimeout());
528
}
529
}
530
}