session.cc 15.3 KB
Newer Older
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15
// 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.

// $Id$
16

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

20
#include <stdint.h>
21

22 23 24 25 26 27
// 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.
28
#include <unistd.h>             // for some IPC/network system calls
29 30
#include <asio.hpp>
#include <asio/error_code.hpp>
31
#include <asio/deadline_timer.hpp>
32 33
#include <asio/system_error.hpp>

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

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

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

46 47
#include <exceptions/exceptions.h>

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

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

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

Jelte Jansen's avatar
Jelte Jansen committed
59
namespace {
60 61 62 63 64
/// \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
65 66 67
setResult(boost::optional<asio::error_code>* result,
          const asio::error_code& err)
{
68 69
    result->reset(err);
}
Jelte Jansen's avatar
Jelte Jansen committed
70 71 72 73
}

namespace isc {
namespace cc {
74

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

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

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

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

    // 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;
118
};
119

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

void
JINMEI Tatuya's avatar
JINMEI Tatuya committed
135
SessionImpl::disconnect() {
136 137
    socket_.close();
    data_length_ = 0;
138 139 140
}

void
JINMEI Tatuya's avatar
JINMEI Tatuya committed
141
SessionImpl::writeData(const void* data, size_t datalen) {
142
    try {
143
        asio::write(socket_, asio::buffer(data, datalen));
144 145
    } catch (const asio::system_error& asio_ex) {
        isc_throw(SessionError, "ASIO write failed: " << asio_ex.what());
146
    }
147 148
}

149
size_t
JINMEI Tatuya's avatar
JINMEI Tatuya committed
150
SessionImpl::readDataLength() {
151 152 153 154 155 156 157 158 159 160 161 162 163 164
    size_t ret_len = data_length_;
    
    if (ret_len == 0) {
        readData(&data_length_, sizeof(data_length_));
        if (data_length_ == 0) {
            isc_throw(SessionError, "ASIO read: data length is not ready");
        }
        ret_len = ntohl(data_length_);
    }

    data_length_ = 0;
    return (ret_len);
}

165
void
166
SessionImpl::readData(void* data, size_t datalen) {
167 168 169
    boost::optional<asio::error_code> read_result;
    boost::optional<asio::error_code> timer_result;

170
    try {
171
        asio::async_read(socket_, asio::buffer(data, datalen),
172
                         boost::bind(&setResult, &read_result, _1));
173 174
        asio::deadline_timer timer(socket_.io_service());
    
175 176
        if (getTimeout() != 0) {
            timer.expires_from_now(boost::posix_time::milliseconds(getTimeout()));
177
            timer.async_wait(boost::bind(&setResult, &timer_result, _1));
178
        }
179

180 181 182 183
        // 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
184 185
        while (!read_result && !timer_result) {
            socket_.io_service().run_one();
Jelte Jansen's avatar
Jelte Jansen committed
186 187

            // Don't cancel the timer if we haven't set it
188
            if (read_result && getTimeout() != 0) {
189
                timer.cancel();
190 191 192
                while (!timer_result) {
                    socket_.io_service().run_one();
                }
193 194
            } else if (timer_result) {
                socket_.cancel();
195 196 197
                while (!read_result) {
                    socket_.io_service().run_one();
                }
198 199
            }
        }
200

201 202
        // asio::error_code evaluates to false if there was no error
        if (*read_result) {
203 204 205 206 207 208 209 210
            if (*read_result == asio::error::operation_aborted) {
                isc_throw(SessionTimeout,
                          "Timeout while reading data from cc session");
            } else {
                isc_throw(SessionError,
                          "Error while reading data from cc session: " <<
                          read_result->message());
            }
211
        }
212
    } catch (const asio::system_error& asio_ex) {
213 214
        // to hide boost specific exceptions, we catch them explicitly
        // and convert it to SessionError.
215
        isc_throw(SessionError, "ASIO read failed: " << asio_ex.what());
216 217 218 219
    }
}

void
JINMEI Tatuya's avatar
JINMEI Tatuya committed
220
SessionImpl::startRead(boost::function<void()> user_handler) {
221 222
    data_length_ = 0;
    user_handler_ = user_handler;
223 224 225 226 227
    asio::async_read(socket_, asio::buffer(&data_length_,
                                           sizeof(data_length_)),
                     boost::bind(&SessionImpl::internalRead, this,
                                 asio::placeholders::error,
                                 asio::placeholders::bytes_transferred));
228 229 230
}

void
JINMEI Tatuya's avatar
JINMEI Tatuya committed
231
SessionImpl::internalRead(const asio::error_code& error,
232
                          size_t bytes_transferred)
233
{
234 235 236 237 238 239 240 241 242 243 244 245
    if (!error) {
        assert(bytes_transferred == sizeof(data_length_));
        data_length_ = ntohl(data_length_);
        if (data_length_ == 0) {
            isc_throw(SessionError, "Invalid message length (0)");
        }
        user_handler_();
    } else {
        isc_throw(SessionError, "asynchronous read failed");
    }
}

JINMEI Tatuya's avatar
JINMEI Tatuya committed
246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261
Session::Session(io_service& io_service) : impl_(new SessionImpl(io_service))
{}

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

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

void
Session::startRead(boost::function<void()> read_callback) {
    impl_->startRead(read_callback);
}
262 263

namespace {                     // maybe unnecessary.
264 265
// This is a helper class to make the establish() method (below) exception-safe
// with the RAII approach.
266
class SessionHolder {
267
public:
268 269 270 271 272 273 274 275 276
    SessionHolder(SessionImpl* obj) : impl_obj_(obj) {}
    ~SessionHolder()
    {
        if (impl_obj_ != NULL) {
            impl_obj_->disconnect();
        }
    }
    void clear() { impl_obj_ = NULL; }
    SessionImpl* impl_obj_;
277
};
278 279
}

280
void
281
Session::establish(const char* socket_file) {
282 283 284 285 286 287 288 289
    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);
JINMEI Tatuya's avatar
JINMEI Tatuya committed
290

291 292 293 294
    // once established, encapsulate the implementation object so that we
    // can safely release the internal resource when exception happens
    // below.
    SessionHolder session_holder(impl_);
295 296 297 298

    //
    // send a request for our local name, and wait for a response
    //
299
    ElementPtr get_lname_msg =
300
        Element::fromJSON("{ \"type\": \"getlname\" }");
301 302
    sendmsg(get_lname_msg);

303 304
    ElementPtr routing, msg;
    recvmsg(routing, msg, false);
305

306 307
    impl_->lname_ = msg->get("lname")->stringValue();

308 309
    // At this point there's no risk of resource leak.
    session_holder.clear();
310 311 312
}

//
JINMEI Tatuya's avatar
JINMEI Tatuya committed
313 314
// Convert to wire format and send this via the stream socket with its length
// prefix.
315 316
//
void
317
Session::sendmsg(ElementPtr& msg) {
318
    std::string header_wire = msg->toWire();
319
    unsigned int length = 2 + header_wire.length();
320
    unsigned int length_net = htonl(length);
321 322
    unsigned short header_length = header_wire.length();
    unsigned short header_length_net = htons(header_length);
323

324 325 326
    impl_->writeData(&length_net, sizeof(length_net));
    impl_->writeData(&header_length_net, sizeof(header_length_net));
    impl_->writeData(header_wire.data(), header_length);
327 328
}

329
void
330
Session::sendmsg(ElementPtr& env, ElementPtr& msg) {
331 332
    std::string header_wire = env->toWire();
    std::string body_wire = msg->toWire();
333 334 335 336 337
    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);

338 339 340 341
    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());
342 343
}

344
bool
Jelte Jansen's avatar
Jelte Jansen committed
345 346 347
Session::recvmsg(ElementPtr& msg, bool nonblock, int seq) {
    ElementPtr l_env;
    return recvmsg(l_env, msg, nonblock, seq);
348 349
}

350
bool
Jelte Jansen's avatar
Jelte Jansen committed
351 352
Session::recvmsg(ElementPtr& env, ElementPtr& msg,
                 bool nonblock, int seq) {
353
    size_t length = impl_->readDataLength();
Jelte Jansen's avatar
Jelte Jansen committed
354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373
    ElementPtr l_env, l_msg;
    if (hasQueuedMsgs()) {
        ElementPtr q_el;
        for (int i = 0; i < impl_->queue_->size(); i++) {
            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);
                   return true;
            }
        }
    }
    
374 375
    unsigned short header_length_net;
    impl_->readData(&header_length_net, sizeof(header_length_net));
376 377

    unsigned short header_length = ntohs(header_length_net);
378 379 380 381
    if (header_length > length || length < 2) {
        isc_throw(SessionError, "Length parameters invalid: total=" << length
                  << ", header=" << header_length);
    }
382 383 384

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

388
    std::string header_wire = std::string(&buffer[0], header_length);
389 390
    std::string body_wire = std::string(&buffer[0] + header_length,
                                        length - header_length);
391 392
    std::stringstream header_wire_stream;
    header_wire_stream << header_wire;
Jelte Jansen's avatar
Jelte Jansen committed
393
    l_env = Element::fromWire(header_wire_stream, header_length);
394
    
395 396
    std::stringstream body_wire_stream;
    body_wire_stream << body_wire;
Jelte Jansen's avatar
Jelte Jansen committed
397 398 399 400 401 402 403 404 405 406 407 408
    l_msg = Element::fromWire(body_wire_stream, length - header_length);
    if ((seq == -1 &&
         !l_env->contains("reply")
        ) || (
         l_env->contains("reply") &&
         l_env->get("reply")->intValue() == seq
        )
       ) {
        env = l_env;
        msg = l_msg;
        return true;
    } else {
409
        ElementPtr q_el = Element::createList();
Jelte Jansen's avatar
Jelte Jansen committed
410 411 412 413 414
        q_el->add(l_env);
        q_el->add(l_msg);
        impl_->queue_->add(q_el);
        return recvmsg(env, msg, nonblock, seq);
    }
415 416 417
    // XXXMLG handle non-block here, and return false for short reads
}

418
void
419
Session::subscribe(std::string group, std::string instance) {
420
    ElementPtr env = Element::createMap();
421 422 423 424 425 426 427 428 429

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

    sendmsg(env);
}

void
430
Session::unsubscribe(std::string group, std::string instance) {
431
    ElementPtr env = Element::createMap();
432 433 434 435 436 437 438 439

    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
440
int
441
Session::group_sendmsg(ElementPtr msg, std::string group,
442
                       std::string instance, std::string to)
443
{
444
    ElementPtr env = Element::createMap();
Jelte Jansen's avatar
Jelte Jansen committed
445
    long int nseq = ++impl_->sequence_;
Jelte Jansen's avatar
Jelte Jansen committed
446
    
447
    env->set("type", Element::create("send"));
448
    env->set("from", Element::create(impl_->lname_));
449 450 451
    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
452
    env->set("seq", Element::create(nseq));
453
    //env->set("msg", Element::create(msg->toWire()));
454

455
    sendmsg(env, msg);
Jelte Jansen's avatar
Jelte Jansen committed
456
    return nseq;
457 458 459
}

bool
460
Session::group_recvmsg(ElementPtr& envelope, ElementPtr& msg,
Jelte Jansen's avatar
Jelte Jansen committed
461
                       bool nonblock, int seq)
462
{
Jelte Jansen's avatar
Jelte Jansen committed
463
    return (recvmsg(envelope, msg, nonblock, seq));
464 465
}

Jelte Jansen's avatar
Jelte Jansen committed
466
int
467
Session::reply(ElementPtr& envelope, ElementPtr& newmsg) {
468
    ElementPtr env = Element::createMap();
Jelte Jansen's avatar
Jelte Jansen committed
469
    long int nseq = ++impl_->sequence_;
Jelte Jansen's avatar
Jelte Jansen committed
470
    
471
    env->set("type", Element::create("send"));
472
    env->set("from", Element::create(impl_->lname_));
473 474 475
    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
476
    env->set("seq", Element::create(nseq));
477
    env->set("reply", Element::create(envelope->get("seq")->intValue()));
478

479
    sendmsg(env, newmsg);
480

Jelte Jansen's avatar
Jelte Jansen committed
481
    return nseq;
482
}
Jelte Jansen's avatar
Jelte Jansen committed
483 484

bool
485
Session::hasQueuedMsgs() {
Jelte Jansen's avatar
Jelte Jansen committed
486 487 488
    return (impl_->queue_->size() > 0);
}

489 490 491 492 493 494
void
Session::setTimeout(size_t milliseconds) {
    impl_->setTimeout(milliseconds);
}

size_t
495
Session::getTimeout() const {
Jelte Jansen's avatar
Jelte Jansen committed
496
    return (impl_->getTimeout());
497
}
498
}
499
}