00001
00002
00003
00004
00005
00006
00007
00008
00009
00010
00011
00012
00013
00014
00015
00016
00017 #ifdef HAVE_CONFIG_H
00018 # include <dtn-config.h>
00019 #endif
00020
00021 #include <sys/poll.h>
00022 #include <stdlib.h>
00023
00024 #include <oasys/io/FileUtils.h>
00025 #include <oasys/util/OptParser.h>
00026 #include <oasys/util/HexDumpBuffer.h>
00027
00028 #include "SerialConvergenceLayer.h"
00029 #include "bundling/BundleDaemon.h"
00030 #include "contacts/ContactManager.h"
00031
00032 namespace dtn {
00033
00034 SerialConvergenceLayer::SerialLinkParams
00035 SerialConvergenceLayer::default_link_params_(true);
00036
00037
00038 SerialConvergenceLayer::SerialLinkParams::SerialLinkParams(bool init_defaults)
00039 : StreamLinkParams(init_defaults),
00040 hexdump_(false),
00041 initstr_(""),
00042 ispeed_(19200),
00043 ospeed_(19200),
00044 sync_interval_(1000)
00045 {
00046 }
00047
00048
00049 SerialConvergenceLayer::SerialConvergenceLayer()
00050 : StreamConvergenceLayer("SerialConvergenceLayer", "serial",
00051 SERIALCL_VERSION)
00052 {
00053 }
00054
00055
00056 ConnectionConvergenceLayer::LinkParams*
00057 SerialConvergenceLayer::new_link_params()
00058 {
00059 return new SerialLinkParams(default_link_params_);
00060 }
00061
00062
00063 bool
00064 SerialConvergenceLayer::parse_link_params(LinkParams* lparams,
00065 int argc, const char** argv,
00066 const char** invalidp)
00067 {
00068 SerialLinkParams* params = dynamic_cast<SerialLinkParams*>(lparams);
00069 ASSERT(params != NULL);
00070
00071 oasys::OptParser p;
00072
00073 p.addopt(new oasys::BoolOpt("hexdump", ¶ms->hexdump_));
00074 p.addopt(new oasys::StringOpt("initstr", ¶ms->initstr_));
00075 p.addopt(new oasys::UIntOpt("sync_interval", ¶ms->sync_interval_));
00076
00077 int count = p.parse_and_shift(argc, argv, invalidp);
00078 if (count == -1) {
00079 return false;
00080 }
00081 argc -= count;
00082
00083
00084 return StreamConvergenceLayer::parse_link_params(lparams, argc, argv,
00085 invalidp);
00086 }
00087
00088
00089 void
00090 SerialConvergenceLayer::dump_link(const LinkRef& link, oasys::StringBuffer* buf)
00091 {
00092 ASSERT(link != NULL);
00093 ASSERT(!link->isdeleted());
00094 ASSERT(link->cl_info() != NULL);
00095
00096 StreamConvergenceLayer::dump_link(link, buf);
00097
00098 SerialLinkParams* params = dynamic_cast<SerialLinkParams*>(link->cl_info());
00099 ASSERT(params != NULL);
00100
00101 buf->appendf("initstr: %s\n", params->initstr_.c_str());
00102 }
00103
00104
00105 bool
00106 SerialConvergenceLayer::set_link_defaults(int argc, const char* argv[],
00107 const char** invalidp)
00108 {
00109 return parse_link_params(&default_link_params_, argc, argv, invalidp);
00110 }
00111
00112
00113 bool
00114 SerialConvergenceLayer::parse_nexthop(const LinkRef& link, LinkParams* lparams)
00115 {
00116 SerialLinkParams* params = dynamic_cast<SerialLinkParams*>(lparams);
00117 ASSERT(params != NULL);
00118
00119 if (! oasys::FileUtils::readable(link->nexthop()))
00120 {
00121 log_warn("can't read tty device file %s", link->nexthop());
00122 return false;
00123 }
00124
00125 return true;
00126 }
00127
00128
00129 CLConnection*
00130 SerialConvergenceLayer::new_connection(const LinkRef& link, LinkParams* p)
00131 {
00132 SerialLinkParams* params = dynamic_cast<SerialLinkParams*>(p);
00133 ASSERT(params != NULL);
00134 return new Connection(this, link, params);
00135 }
00136
00137
00138 SerialConvergenceLayer::Connection::Connection(SerialConvergenceLayer* cl,
00139 const LinkRef& link,
00140 SerialLinkParams* params)
00141 : StreamConvergenceLayer::Connection("SerialConvergenceLayer::Connection",
00142 cl->logpath(), cl, params,
00143 true )
00144 {
00145 logpathf("%s/conn/%p", cl->logpath(), this);
00146
00147
00148 set_nexthop(link->nexthop());
00149
00150
00151 tty_ = new oasys::TTY(logpath_);
00152 tty_->logpathf("%s/tty", logpath_);
00153
00154 synced_ = false;
00155 }
00156
00157
00158 SerialConvergenceLayer::Connection::~Connection()
00159 {
00160 delete tty_;
00161 }
00162
00163
00164 void
00165 SerialConvergenceLayer::Connection::serialize(oasys::SerializeAction *a)
00166 {
00167
00168 (void)a;
00169 }
00170
00171
00172 void
00173 SerialConvergenceLayer::Connection::initialize_pollfds()
00174 {
00175
00176
00177 const LinkRef& link = contact_->link();
00178
00179
00180
00181 if (! cl_->parse_nexthop(link, params_)) {
00182 log_info("can't resolve nexthop address '%s'", link->nexthop());
00183 break_contact(ContactEvent::BROKEN);
00184 return;
00185 }
00186
00187
00188 int ret = tty_->open(link->nexthop(), O_RDWR | O_NOCTTY);
00189 if (ret == -1) {
00190 log_info("opening %s failed... %s", link->nexthop(), strerror(errno));
00191 break_contact(ContactEvent::BROKEN);
00192 return;
00193 }
00194
00195 log_debug("opened %s", link->nexthop());
00196 if (!tty_->isatty()) {
00197 log_err("%s is not a TTY", link->nexthop());
00198 break_contact(ContactEvent::BROKEN);
00199 return;
00200 }
00201
00202 log_debug("setting tty parameters...");
00203 tty_->tcgetattr();
00204 tty_->cfmakeraw();
00205 tty_->cfsetispeed(serial_lparams()->ispeed_);
00206 tty_->cfsetospeed(serial_lparams()->ospeed_);
00207 tty_->tcflush(TCIOFLUSH);
00208 tty_->tcsetattr(TCSANOW);
00209 tty_->set_nonblocking(true);
00210
00211 tty_pollfd_ = &pollfds_[0];
00212 num_pollfds_ = 1;
00213
00214 tty_pollfd_->fd = tty_->fd();
00215 tty_pollfd_->events = POLLIN;
00216
00217 poll_timeout_ = serial_lparams()->sync_interval_;
00218 }
00219
00220
00221 void
00222 SerialConvergenceLayer::Connection::connect()
00223 {
00224
00225
00226 ::gettimeofday(&data_rcvd_, 0);
00227
00228
00229 SerialLinkParams* params = serial_lparams();
00230 size_t initstr_len = params->initstr_.length();
00231 if (initstr_len != 0) {
00232 log_debug("copying initialization string \"%s\"",
00233 params->initstr_.c_str());
00234
00235
00236 sendbuf_.reserve(initstr_len);
00237 memcpy(sendbuf_.end(), params->initstr_.data(), initstr_len);
00238 sendbuf_.fill(initstr_len);
00239 }
00240
00241
00242 send_sync();
00243 }
00244
00245
00246 void
00247 SerialConvergenceLayer::Connection::disconnect()
00248 {
00249 if (tty_->fd() != -1) {
00250 tty_->close();
00251 }
00252 }
00253
00254
00255 void
00256 SerialConvergenceLayer::Connection::send_sync()
00257 {
00258
00259
00260 if (sendbuf_.tailbytes() == 0) {
00261 log_debug("send_sync: "
00262 "send buffer has %zu bytes queued, suppressing sync",
00263 sendbuf_.fullbytes());
00264 return;
00265 }
00266 ASSERT(sendbuf_.tailbytes() > 0);
00267
00268 *(sendbuf_.end()) = SYNC;
00269 sendbuf_.fill(1);
00270
00271 send_data();
00272 }
00273
00274
00275 void
00276 SerialConvergenceLayer::Connection::handle_poll_timeout()
00277 {
00278 if (!synced_) {
00279 struct timeval now;
00280 u_int elapsed;
00281 SerialLinkParams* params = serial_lparams();
00282
00283 ::gettimeofday(&now, 0);
00284
00285
00286
00287 elapsed = TIMEVAL_DIFF_MSEC(now, data_rcvd_);
00288 if (elapsed > params->data_timeout_) {
00289 log_info("handle_poll_timeout: no data heard for %d msecs "
00290 "(data_rcvd %u.%u, now %u.%u, data_timeout %d) "
00291 "-- closing contact",
00292 elapsed,
00293 (u_int)data_rcvd_.tv_sec, (u_int)data_rcvd_.tv_usec,
00294 (u_int)now.tv_sec, (u_int)now.tv_usec,
00295 params->data_timeout_);
00296
00297 break_contact(ContactEvent::BROKEN);
00298 return;
00299 }
00300
00301 log_debug("handle_poll_timeout: sending another sync byte");
00302 send_sync();
00303 } else {
00304
00305 StreamConvergenceLayer::Connection::handle_poll_timeout();
00306 }
00307 }
00308
00309
00310 void
00311 SerialConvergenceLayer::Connection::handle_poll_activity()
00312 {
00313 if (tty_pollfd_->revents & POLLHUP) {
00314 log_info("tty closed connection -- returned POLLHUP");
00315 break_contact(ContactEvent::BROKEN);
00316 return;
00317 }
00318
00319 if (tty_pollfd_->revents & POLLERR) {
00320 log_info("error condition on tty -- returned POLLERR");
00321 break_contact(ContactEvent::BROKEN);
00322 return;
00323 }
00324
00325
00326
00327
00328 if (tty_pollfd_->revents & POLLOUT)
00329 {
00330 log_debug("poll returned write ready, clearing POLLOUT bit");
00331 tty_pollfd_->events &= ~POLLOUT;
00332 send_data();
00333 }
00334
00335
00336 if (contact_broken_)
00337 {
00338 return;
00339 }
00340
00341
00342 if (tty_pollfd_->revents & POLLIN) {
00343 recv_data();
00344 process_data();
00345
00346
00347
00348 if (recvbuf_.tailbytes() == 0) {
00349 log_err("process_data left no space in recvbuf!!");
00350 }
00351
00352 if (contact_up_ && ! contact_broken_) {
00353 check_keepalive();
00354 }
00355 }
00356 }
00357
00358
00359 void
00360 SerialConvergenceLayer::Connection::send_data()
00361 {
00362
00363
00364
00365 ASSERT(! contact_broken_);
00366
00367 u_int towrite = sendbuf_.fullbytes();
00368 if (params_->test_write_limit_ != 0) {
00369 towrite = std::min(towrite, params_->test_write_limit_);
00370 }
00371
00372 log_debug("send_data: trying to drain %u bytes from send buffer...",
00373 towrite);
00374 ASSERT(towrite > 0);
00375
00376 int cc = tty_->write(sendbuf_.start(), towrite);
00377 if (cc > 0) {
00378 log_debug("send_data: wrote %d/%zu bytes from send buffer",
00379 cc, sendbuf_.fullbytes());
00380 if (serial_lparams()->hexdump_) {
00381 oasys::HexDumpBuffer hex;
00382 hex.append((u_char*)sendbuf_.start(), cc);
00383 log_multiline(oasys::LOG_ALWAYS, hex.hexify().c_str());
00384 }
00385
00386 sendbuf_.consume(cc);
00387
00388 if (sendbuf_.fullbytes() != 0) {
00389 log_debug("send_data: incomplete write, setting POLLOUT bit");
00390 tty_pollfd_->events |= POLLOUT;
00391
00392 } else {
00393 if (tty_pollfd_->events & POLLOUT) {
00394 log_debug("send_data: drained buffer, clearing POLLOUT bit");
00395 tty_pollfd_->events &= ~POLLOUT;
00396 }
00397 }
00398 } else if (errno == EWOULDBLOCK) {
00399 log_debug("send_data: write returned EWOULDBLOCK, setting POLLOUT bit");
00400 tty_pollfd_->events |= POLLOUT;
00401
00402 } else {
00403 log_info("send_data: remote connection unexpectedly closed: %s",
00404 strerror(errno));
00405 break_contact(ContactEvent::BROKEN);
00406 }
00407 }
00408
00409
00410 void
00411 SerialConvergenceLayer::Connection::recv_data()
00412 {
00413
00414
00415
00416 ASSERT(! contact_broken_);
00417
00418
00419 if (recvbuf_.tailbytes() == 0) {
00420 log_err("no space in receive buffer to accept data!!!");
00421 return;
00422 }
00423
00424 if (params_->test_read_delay_ != 0) {
00425 log_debug("recv_data: sleeping for test_read_delay msecs %u",
00426 params_->test_read_delay_);
00427
00428 usleep(params_->test_read_delay_ * 1000);
00429 }
00430
00431 u_int toread = recvbuf_.tailbytes();
00432 if (params_->test_read_limit_ != 0) {
00433 toread = std::min(toread, params_->test_read_limit_);
00434 }
00435
00436 log_debug("recv_data: draining up to %u bytes into recv buffer...", toread);
00437 int cc = tty_->read(recvbuf_.end(), toread);
00438 if (cc < 1) {
00439 log_info("remote connection unexpectedly closed");
00440 break_contact(ContactEvent::BROKEN);
00441 return;
00442 }
00443
00444 log_debug("recv_data: read %d bytes, rcvbuf has %zu bytes",
00445 cc, recvbuf_.fullbytes());
00446 if (serial_lparams()->hexdump_) {
00447 oasys::HexDumpBuffer hex;
00448 hex.append((u_char*)recvbuf_.end(), cc);
00449 log_multiline(oasys::LOG_ALWAYS, hex.hexify().c_str());
00450 }
00451 recvbuf_.fill(cc);
00452
00453
00454
00455 if (! contact_initiated_) {
00456 initiate_contact();
00457 }
00458
00459
00460 if (! synced_)
00461 {
00462 while ((recvbuf_.fullbytes() != 0) &&
00463 (*(u_char*)recvbuf_.start() == SYNC))
00464 {
00465 log_debug("got a sync byte... ignoring");
00466 recvbuf_.consume(1);
00467 }
00468
00469
00470
00471 if (recvbuf_.fullbytes() != 0)
00472 {
00473 log_debug("done reading sync bytes, clearing synced flag");
00474 synced_ = true;
00475 }
00476
00477
00478 SerialLinkParams* params = serial_lparams();
00479 poll_timeout_ = params->data_timeout_;
00480
00481 if (params->keepalive_interval_ != 0 &&
00482 (params->keepalive_interval_ * 1000) < params->data_timeout_)
00483 {
00484 poll_timeout_ = params->keepalive_interval_ * 1000;
00485 }
00486 }
00487 }
00488
00489 }