Attn: Mike Belshe

net/socket/sctp_client_socket_libevent.cc

+// Copyright (c) 2011 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#include "net/socket/sctp_client_socket.h"
+#include "net/socket/sctp_support.h"
+
+#include <errno.h>
+#include <fcntl.h>
+#include <netdb.h>
+#include <sys/socket.h>
+#include <sys/types.h>
+#include <netinet/sctp.h>
+#include <netinet/tcp.h>
+#if defined(OS_POSIX)
+#include <netinet/in.h>
+#include <arpa/inet.h>
+#endif
+
+#include "base/eintr_wrapper.h"
+#include "base/logging.h"
+#include "base/message_loop.h"
+#include "base/metrics/stats_counters.h"
+#include "base/string_util.h"
+#include "net/base/address_list_net_log_param.h"
+#include "net/base/connection_type_histograms.h"
+#include "net/base/io_buffer.h"
+#include "net/base/net_errors.h"
+#include "net/base/net_log.h"
+#include "net/base/net_util.h"
+#include "net/base/network_change_notifier.h"
+#if defined(USE_SYSTEM_LIBEVENT)
+#include <event.h>
+#else
+#include "third_party/libevent/event.h"
+#endif
+
+namespace net {
+
+namespace {
+
+const int kInvalidSocket = -1;
+
+// These values need to be reviewed and set based on experimental results.
+// TODO(jtl): Need to come up with a rationale for choosing this value.
+const int kSctpMaxBurst = 4096;     // Arbitrary value, but high enough so
+                                    // that maxburst has no effect.
+const uint16 kNumberOfSctpStreamsToRequest = 50;
+
+// DisableNagle turns off buffering in the kernel. By default, SCTP sockets will
+// wait up to 200ms for more data to complete a packet before transmitting.
+// After calling this function, the kernel will not wait. See SCTP_NODELAY in
+// `man 7 sctp`.
+void DisableNagle(int fd) {
+  int on = 1;
+  if (setsockopt(fd, IPPROTO_SCTP, SCTP_NODELAY, &on, sizeof(on)) < 0)
+    PLOG(ERROR) << "Failed to set SCTP_NODELAY on fd: " << fd;
+  return;
+}
+
+// This is SCTP's equivalent to TCP's keepalive interval
+void SetSCTPHeartbeatInterval(int fd) {
+  struct sctp_paddrparams params;
+  socklen_t params_len = sizeof(params);
+  memset(&params, 0, params_len);
+  if (getsockopt(fd, IPPROTO_SCTP, SCTP_PEER_ADDR_PARAMS, &params,
+                    &params_len)) {
+    PLOG(ERROR) << "Failed to get SCTP_PEER_ADDR_PARAMS on fd: " << fd;
+    return;
+  }
+  // Increase hearbeat interval to 45 seconds.
+  params.spp_hbinterval = 45000;
+  if (setsockopt(fd, IPPROTO_SCTP, SCTP_PEER_ADDR_PARAMS, &params,
+                 params_len)) {
+    PLOG(ERROR) << "Failed to set SCTP_PEER_ADDR_PARAMS on fd: " << fd;
+    return;
+  }
+  return;
+}
+
+void SetSCTPMaxBurst(int fd) {
+  int maxburst = kSctpMaxBurst;
+  socklen_t maxburst_len = sizeof(maxburst);
+  if (setsockopt(fd, IPPROTO_SCTP, SCTP_MAX_BURST, &maxburst,
+                 maxburst_len) < 0) {
+    PLOG(ERROR) << "Failed to set SCTP_MAX_BURST on fd: " << fd << "\n";
+  }
+  return;
+}
+
+// Set number of outgoing SCTP streams to request
+void SetSCTPInitMsg(int fd) {

[Mike Belshe, 2011/04/06 18:32:53]

I would rename to "SetSCTPInitialStreams"

+  struct sctp_initmsg initmsg;
+  socklen_t initmsg_len = sizeof(initmsg);
+  memset(&initmsg, 0, initmsg_len);
+  if (getsockopt(fd, IPPROTO_SCTP, SCTP_INITMSG, &initmsg, &initmsg_len) < 0) {
+    PLOG(ERROR) << "Failed to get SCTP_INITMSG on fd:" << fd << "\n";
+    return;
+  }
+  initmsg.sinit_num_ostreams = kNumberOfSctpStreamsToRequest;
+  initmsg_len = sizeof(initmsg);
+  if (setsockopt(fd, IPPROTO_SCTP, SCTP_INITMSG, &initmsg, initmsg_len) < 0)
+    PLOG(ERROR) << "Failed to set SCTP_INITMSG on fd:" << fd << "\n";
+

[Mike Belshe, 2011/04/06 18:32:53]

nit: remove lines 104/105

+  return;
+}
+
+// Convert values from <errno.h> to values from "net/base/net_errors.h"
+int MapPosixError(int os_error) {
+  // There are numerous posix error codes, but these are the ones we thus far
+  // find interesting.
+  switch (os_error) {
+    case EAGAIN:
+#if EWOULDBLOCK != EAGAIN
+    case EWOULDBLOCK:
+#endif
+      return ERR_IO_PENDING;
+    case EACCES:
+      return ERR_ACCESS_DENIED;
+    case ENETDOWN:
+      return ERR_INTERNET_DISCONNECTED;
+    case ETIMEDOUT:
+      return ERR_TIMED_OUT;
+    case ECONNRESET:
+    case ENETRESET:  // Related to keep-alive
+    case EPIPE:
+      return ERR_CONNECTION_RESET;
+    case ECONNABORTED:
+      return ERR_CONNECTION_ABORTED;
+    case ECONNREFUSED:
+      return ERR_CONNECTION_REFUSED;
+    case EHOSTUNREACH:
+    case EHOSTDOWN:
+    case ENETUNREACH:
+      return ERR_ADDRESS_UNREACHABLE;
+    case EADDRNOTAVAIL:
+      return ERR_ADDRESS_INVALID;
+    case 0:
+      return OK;
+    default:
+      LOG(WARNING) << "Unknown error " << os_error
+                   << " mapped to net::ERR_FAILED";
+      return ERR_FAILED;
+  }
+}
+
+int MapConnectError(int os_error) {
+  switch (os_error) {
+    case EACCES:
+      return ERR_NETWORK_ACCESS_DENIED;
+    case ETIMEDOUT:
+      return ERR_CONNECTION_TIMED_OUT;
+    default: {
+      int net_error = MapPosixError(os_error);
+      if (net_error == ERR_FAILED)
+        return ERR_CONNECTION_FAILED;  // More specific than ERR_FAILED.
+
+      // Give a more specific error when the user is offline.
+      if (net_error == ERR_ADDRESS_UNREACHABLE &&
+          NetworkChangeNotifier::IsOffline()) {
+        return ERR_INTERNET_DISCONNECTED;
+      }
+      return net_error;
+    }
+  }
+}
+
+}  // namespace
+
+//-----------------------------------------------------------------------------
+
+SCTPClientSocketLibevent::SCTPClientSocketLibevent(
+    const AddressList& addresses,
+    net::NetLog* net_log,
+    const net::NetLog::Source& source)
+    : socket_(kInvalidSocket),
+      addresses_(addresses),
+      current_ai_(NULL),
+      using_sctp_control_stream_(sctp_control_stream_enabled()),
+      max_sctp_streams_(0),
+      read_watcher_(this),

[Mike Belshe, 2011/04/06 18:32:53]

nit: missed 
  read_buf_len_(0)
  write_buf_len_(0)

+      write_watcher_(this),
+      read_callback_(NULL),
+      write_callback_(NULL),
+      next_connect_state_(CONNECT_STATE_NONE),
+      connect_os_error_(0),
+      net_log_(BoundNetLog::Make(net_log, NetLog::SOURCE_SOCKET)),
+      previously_disconnected_(false) {
+  scoped_refptr<NetLog::EventParameters> params;
+  if (source.is_valid())
+    params = new NetLogSourceParameter("source_dependency", source);
+  net_log_.BeginEvent(NetLog::TYPE_SOCKET_ALIVE, params);
+}
+
+SCTPClientSocketLibevent::~SCTPClientSocketLibevent() {
+  Disconnect();
+  net_log_.EndEvent(NetLog::TYPE_SOCKET_ALIVE, NULL);
+}
+
+void SCTPClientSocketLibevent::AdoptSocket(int socket) {
+  DCHECK_EQ(socket_, kInvalidSocket);
+  socket_ = socket;
+  int error = SetupSocket();
+  DCHECK_EQ(0, error);
+  // This is to make GetPeerAddress work. It's up to the test that is calling
+  // this function to ensure that address_ contains a reasonable address for
+  // this socket. (i.e. at least match IPv4 vs IPv6!).
+  current_ai_ = addresses_.head();
+  use_history_.set_was_ever_connected();
+}
+
+int SCTPClientSocketLibevent::Connect(CompletionCallback* callback) {
+  DCHECK(CalledOnValidThread());
+
+  // If already connected, then just return OK.
+  if (socket_ != kInvalidSocket)
+    return OK;
+
+  static base::StatsCounter connects("sctp.connect");
+  connects.Increment();
+
+  DCHECK(!waiting_connect());
+
+  net_log_.BeginEvent(
+      NetLog::TYPE_SCTP_CONNECT,
+      make_scoped_refptr(new AddressListNetLogParam(addresses_)));
+
+  // We will try to connect to each address in addresses_. Start with the
+  // first one in the list.
+  next_connect_state_ = CONNECT_STATE_CONNECT;
+  current_ai_ = addresses_.head();
+
+  int rv = DoConnectLoop(OK);
+  if (rv == ERR_IO_PENDING) {
+    // Synchronous operation not supported.
+    DCHECK(callback);
+    write_callback_ = callback;
+  } else {
+    LogConnectCompletion(rv);
+  }
+
+  return rv;
+}
+
+int SCTPClientSocketLibevent::DoConnectLoop(int result) {
+  DCHECK_NE(next_connect_state_, CONNECT_STATE_NONE);
+
+  int rv = result;
+  do {
+    ConnectState state = next_connect_state_;
+    next_connect_state_ = CONNECT_STATE_NONE;
+    switch (state) {
+      case CONNECT_STATE_CONNECT:
+        DCHECK_EQ(OK, rv);
+        rv = DoConnect();
+        break;
+      case CONNECT_STATE_CONNECT_COMPLETE:
+        rv = DoConnectComplete(rv);
+        break;
+      default:
+        LOG(DFATAL) << "bad state";
+        rv = ERR_UNEXPECTED;
+        break;
+    }
+  } while (rv != ERR_IO_PENDING && next_connect_state_ != CONNECT_STATE_NONE);
+
+  return rv;
+}
+
+int SCTPClientSocketLibevent::DoConnect() {
+  DCHECK(current_ai_);
+
+  DCHECK_EQ(0, connect_os_error_);
+
+  if (previously_disconnected_) {
+    use_history_.Reset();
+    previously_disconnected_ = false;
+  }
+
+  net_log_.BeginEvent(NetLog::TYPE_SCTP_CONNECT_ATTEMPT,
+                      make_scoped_refptr(new NetLogStringParameter(
+                          "address", NetAddressToStringWithPort(current_ai_))));
+
+  next_connect_state_ = CONNECT_STATE_CONNECT_COMPLETE;
+
+  // Create a non-blocking socket.
+  connect_os_error_ = CreateSocket(current_ai_);
+  if (connect_os_error_)
+    return MapPosixError(connect_os_error_);
+
+  // Connect the socket.
+  if (!HANDLE_EINTR(connect(socket_, current_ai_->ai_addr,
+                            static_cast<int>(current_ai_->ai_addrlen)))) {
+    // Connected without waiting!
+    return OK;
+  }
+
+  // Check if the connect() failed synchronously.
+  connect_os_error_ = errno;
+  if (connect_os_error_ != EINPROGRESS)
+    return MapConnectError(connect_os_error_);
+
+  // Otherwise the connect() is going to complete asynchronously, so watch
+  // for its completion.
+  if (!MessageLoopForIO::current()->WatchFileDescriptor(
+          socket_, true, MessageLoopForIO::WATCH_WRITE, &write_socket_watcher_,
+          &write_watcher_)) {
+    connect_os_error_ = errno;
+    DVLOG(1) << "WatchFileDescriptor failed: " << connect_os_error_;
+    return MapPosixError(connect_os_error_);
+  }
+
+  return ERR_IO_PENDING;
+}
+
+int SCTPClientSocketLibevent::DoConnectComplete(int result) {
+  // Log the end of this attempt (and any OS error it threw).
+  int os_error = connect_os_error_;
+  connect_os_error_ = 0;
+  scoped_refptr<NetLog::EventParameters> params;
+  if (result != OK)
+    params = new NetLogIntegerParameter("os_error", os_error);
+  net_log_.EndEvent(NetLog::TYPE_SCTP_CONNECT_ATTEMPT, params);
+
+  write_socket_watcher_.StopWatchingFileDescriptor();
+
+  if (result == OK) {
+    use_history_.set_was_ever_connected();
+
+    // Get number of allowable outgoing SCTP streams and set max_sctp_streams_.
+    struct sctp_status status;
+    socklen_t status_len = sizeof(status);
+    int rv;
+    memset(&status, 0, status_len);
+    rv = getsockopt(socket_, IPPROTO_SCTP, SCTP_STATUS, &status, &status_len);
+    if (rv < 0 )

[Mike Belshe, 2011/04/06 18:32:53]

This must be a bug - missing curly braces, causes this function to always set rv
= ERR_UNEXPECTED?

+      LOG(ERROR) << "Unable to get status of SCTP socket.";
+      rv = ERR_UNEXPECTED;
+
+    if (status.sstat_state == SCTP_ESTABLISHED) {
+      max_sctp_streams_ = status.sstat_instrms <= status.sstat_outstrms ?
+                          status.sstat_instrms : status.sstat_outstrms;
+      return OK;  // Done!
+    } else if (status.sstat_state == SCTP_COOKIE_WAIT) {
+      // TODO(jtl): Need to figure out how and where to best handle this case.
+      // This is actually the prefered outcome because it potentially saves an
+      // RTT, and it can occur on FreeBSD and Mac OS X.
+      LOG(ERROR) << "Unable to set max_sctp_streams_ until association is "
+                 << "established.";
+      rv = ERR_UNEXPECTED;

[Mike Belshe, 2011/04/06 18:32:53]

the return value is |result| not |rv|, so these aren't working...

+    } else {
+      LOG(ERROR) << "Unexpected association state " << status.sstat_state
+                 << " on connecting socket " << socket_;
+      rv = ERR_UNEXPECTED;
+    }
+  }
+
+  // Close whatever partially connected socket we currently have.
+  DoDisconnect();
+
+  // Try to fall back to the next address in the list.
+  if (current_ai_->ai_next) {
+    next_connect_state_ = CONNECT_STATE_CONNECT;
+    current_ai_ = current_ai_->ai_next;
+    return OK;
+  }
+
+  // Otherwise there is nothing to fall back to, so give up.
+  return result;
+}
+
+void SCTPClientSocketLibevent::Disconnect() {
+  DCHECK(CalledOnValidThread());
+
+  DoDisconnect();
+  current_ai_ = NULL;
+}
+
+void SCTPClientSocketLibevent::DoDisconnect() {
+  if (socket_ == kInvalidSocket)
+    return;
+
+  bool ok = read_socket_watcher_.StopWatchingFileDescriptor();
+  DCHECK(ok);
+  ok = write_socket_watcher_.StopWatchingFileDescriptor();
+  DCHECK(ok);
+  if (HANDLE_EINTR(close(socket_)) < 0)
+    PLOG(ERROR) << "close";
+  socket_ = kInvalidSocket;
+  previously_disconnected_ = true;
+}
+
+bool SCTPClientSocketLibevent::IsConnected() const {
+  DCHECK(CalledOnValidThread());
+
+  if (socket_ == kInvalidSocket || waiting_connect())
+    return false;
+
+  // Check if connection is alive.
+  char c;
+  int rv = HANDLE_EINTR(recv(socket_, &c, 1, MSG_PEEK));
+  if (rv == 0)
+    return false;
+  if (rv == -1 && errno != EAGAIN && errno != EWOULDBLOCK)
+    return false;
+
+  return true;
+}
+
+bool SCTPClientSocketLibevent::IsConnectedAndIdle() const {
+  DCHECK(CalledOnValidThread());
+
+  if (socket_ == kInvalidSocket || waiting_connect())
+    return false;
+
+  // Check if connection is alive and we haven't received any data
+  // unexpectedly.
+  char c;
+  int rv = HANDLE_EINTR(recv(socket_, &c, 1, MSG_PEEK));
+  if (rv >= 0)
+    return false;
+  if (errno != EAGAIN && errno != EWOULDBLOCK)
+    return false;
+
+  return true;
+}
+
+uint16 SCTPClientSocketLibevent::MapSpdyToSctp(uint32 stream_id) {
+  DCHECK_GT(max_sctp_streams_, 0);
+
+  uint16 value = max_sctp_streams_;
+  if (!(stream_id & 0x00000001))

[Mike Belshe, 2011/04/06 18:32:53]

What is this doing?

+    value -= 1;
+
+  return (stream_id - 1) % (value & 0xfffe) + 1;
+}
+
+int SCTPClientSocketLibevent::Read(IOBuffer* buf,
+                                  int buf_len,
+                                  CompletionCallback* callback) {
+  DCHECK(CalledOnValidThread());
+  DCHECK_NE(kInvalidSocket, socket_);
+  DCHECK(!waiting_connect());
+  DCHECK(!read_callback_);
+  // Synchronous operation not supported
+  DCHECK(callback);
+  DCHECK_GT(buf_len, 0);
+
+  int nread = HANDLE_EINTR(read(socket_, buf->data(), buf_len));

[Mike Belshe, 2011/04/06 18:32:53]

I was surprised to see we didn't need to know the stream-id for this.

OK - this a bug for the multi-packet spdy frame reading...  we should at least
add a TODO() that it is known to be broken...

In this current model, it probably needs to call buf->set_sctp_stream_id(xx)?

+  if (nread >= 0) {
+    static base::StatsCounter read_bytes("sctp.read_bytes");
+    read_bytes.Add(nread);
+    if (nread > 0)
+      use_history_.set_was_used_to_convey_data();
+    LogByteTransfer(
+        net_log_, NetLog::TYPE_SOCKET_BYTES_RECEIVED, nread, buf->data());
+    return nread;
+  }
+  if (errno != EAGAIN && errno != EWOULDBLOCK) {
+    DVLOG(1) << "read failed, errno " << errno;
+    return MapPosixError(errno);
+  }
+
+  if (!MessageLoopForIO::current()->WatchFileDescriptor(
+          socket_, true, MessageLoopForIO::WATCH_READ,
+          &read_socket_watcher_, &read_watcher_)) {
+    DVLOG(1) << "WatchFileDescriptor failed on read, errno " << errno;
+    return MapPosixError(errno);
+  }
+
+  read_buf_ = buf;
+  read_buf_len_ = buf_len;
+  read_callback_ = callback;
+  return ERR_IO_PENDING;
+}
+
+int SCTPClientSocketLibevent::Write(IOBuffer* buf,
+                                   int buf_len,
+                                   CompletionCallback* callback) {
+  DCHECK(CalledOnValidThread());
+  DCHECK_NE(kInvalidSocket, socket_);
+  DCHECK(!waiting_connect());
+  DCHECK(!write_callback_);
+  // Synchronous operation not supported
+  DCHECK(callback);
+  DCHECK_GT(buf_len, 0);
+
+  int nwrite = InternalWrite(buf, buf_len);
+  if (nwrite >= 0) {
+    static base::StatsCounter write_bytes("sctp.write_bytes");
+    write_bytes.Add(nwrite);
+    if (nwrite > 0)
+      use_history_.set_was_used_to_convey_data();
+    LogByteTransfer(
+        net_log_, NetLog::TYPE_SOCKET_BYTES_SENT, nwrite, buf->data());
+    return nwrite;
+  }
+  if (errno != EAGAIN && errno != EWOULDBLOCK)
+    return MapPosixError(errno);
+
+  if (!MessageLoopForIO::current()->WatchFileDescriptor(
+          socket_, true, MessageLoopForIO::WATCH_WRITE,
+          &write_socket_watcher_, &write_watcher_)) {
+    DVLOG(1) << "WatchFileDescriptor failed on write, errno " << errno;
+    return MapPosixError(errno);
+  }
+
+  write_buf_ = buf;
+  write_buf_len_ = buf_len;
+  write_callback_ = callback;
+  return ERR_IO_PENDING;
+}
+
+int SCTPClientSocketLibevent::InternalWrite(IOBuffer* buf, int buf_len) {
+  return HANDLE_EINTR(sctp_sendmsg(socket_, buf->data(), buf_len,
+                      NULL, 0, 0, 0, buf->sctp_stream_id(), 0, 0));
+}
+
+bool SCTPClientSocketLibevent::SetReceiveBufferSize(int32 size) {
+  DCHECK(CalledOnValidThread());
+  int rv = setsockopt(socket_, SOL_SOCKET, SO_RCVBUF,
+      reinterpret_cast<const char*>(&size),
+      sizeof(size));
+  DCHECK(!rv) << "Could not set socket receive buffer size: " << errno;
+  return rv == 0;
+}
+
+bool SCTPClientSocketLibevent::SetSendBufferSize(int32 size) {
+  DCHECK(CalledOnValidThread());
+  int rv = setsockopt(socket_, SOL_SOCKET, SO_SNDBUF,
+      reinterpret_cast<const char*>(&size),
+      sizeof(size));
+  DCHECK(!rv) << "Could not set socket send buffer size: " << errno;
+  return rv == 0;
+}
+
+
+int SCTPClientSocketLibevent::CreateSocket(const addrinfo* ai) {
+  socket_ = socket(ai->ai_family, ai->ai_socktype, IPPROTO_SCTP);
+  if (socket_ == kInvalidSocket)
+    return errno;
+  return SetupSocket();
+}
+
+int SCTPClientSocketLibevent::SetupSocket() {
+  // Linux SCTP has a bug which causes retransmissions to go to unconfirmed
+  // addresses.  Since the default behaviour, when calling connect() with an
+  // unbound socket, is to bind the socket to all local addresses, the server
+  // may retransmit packets to some or all of these interfaces even if they are
+  // unreachable.  One work around is to bind the socket to a single local
+  // address prior to calling connect().  An alternative is to use sctp_bindx()
+  // to bind only those addresses that are reachable.
+  // The bug exists in 2.6.31 and persists at least through 2.6.34.  I do not
+  // know about earlier or later versions.
+  // TODO(jtl): Need to implement a way to select a single local IP address to
+  // bind to.
+  struct sockaddr_in sin;
+  sin.sin_family = AF_INET;
+  sin.sin_port = htons(0);    // let system choose ephemeral port number
+  //inet_aton("127.0.0.1", &(sin.sin_addr));

[Mike Belshe, 2011/04/06 18:32:53]

Wow - what a mess!

Not sure what to do to make this code usable?  Maybe working around the bug
should be put into a special function with an ifdef or something...

+  inet_aton("10.1.207.2", &(sin.sin_addr));
+  //inet_aton("128.4.30.27", &(sin.sin_addr));
+  if (bind(socket_, (struct sockaddr*) &sin, sizeof(sin)) < 0) {
+    const int bind_os_error_ = errno;
+    printf("SCTPClientSocketLibevent::SetupSocket: bind() failed with errno = "
+           "%d (%s)\n", errno, strerror(errno));
+    return bind_os_error_;
+  }
+
+  if (SetNonBlocking(socket_)) {
+    const int err = errno;
+    close(socket_);
+    socket_ = kInvalidSocket;
+    return err;
+  }
+
+  // This mirrors the behaviour on Windows. See the comment in
+  // tcp_client_socket_win.cc after searching for "NODELAY".
+  DisableNagle(socket_);  // If DisableNagle fails, we don't care.
+
+  // Heartbeats are SCTP's equivalent to TCP's keepalive, and they're always on.
+  SetSCTPHeartbeatInterval(socket_);
+
+  // TODO(jtl): Make this a command line option.
+  // set max_burst
+  SetSCTPMaxBurst(socket_);
+
+  // Set the number of outgoing sctp streams to request.
+  SetSCTPInitMsg(socket_);
+  return 0;
+}
+
+void SCTPClientSocketLibevent::LogConnectCompletion(int net_error) {
+  if (net_error != OK) {
+    net_log_.EndEventWithNetErrorCode(NetLog::TYPE_SCTP_CONNECT, net_error);
+    return;
+  }
+
+  struct sockaddr_storage source_address;
+  socklen_t addrlen = sizeof(source_address);
+  int rv = getsockname(
+      socket_, reinterpret_cast<struct sockaddr*>(&source_address), &addrlen);
+  if (rv != 0) {
+    PLOG(ERROR) << "getsockname() [rv: " << rv << "] error: ";
+    NOTREACHED();
+    net_log_.EndEventWithNetErrorCode(NetLog::TYPE_SCTP_CONNECT, rv);
+    return;
+  }
+
+  const std::string source_address_str =
+      NetAddressToStringWithPort(
+          reinterpret_cast<const struct sockaddr*>(&source_address),
+          sizeof(source_address));
+  net_log_.EndEvent(NetLog::TYPE_SCTP_CONNECT,
+                    make_scoped_refptr(new NetLogStringParameter(
+                        "source address",
+                        source_address_str)));
+}
+
+void SCTPClientSocketLibevent::DoReadCallback(int rv) {
+  DCHECK_NE(rv, ERR_IO_PENDING);
+  DCHECK(read_callback_);
+
+  // since Run may result in Read being called, clear read_callback_ up front.
+  CompletionCallback* c = read_callback_;
+  read_callback_ = NULL;
+  c->Run(rv);
+}
+
+void SCTPClientSocketLibevent::DoWriteCallback(int rv) {
+  DCHECK_NE(rv, ERR_IO_PENDING);
+  DCHECK(write_callback_);
+
+  // since Run may result in Write being called, clear write_callback_ up front.
+  CompletionCallback* c = write_callback_;
+  write_callback_ = NULL;
+  c->Run(rv);
+}
+
+void SCTPClientSocketLibevent::DidCompleteConnect() {
+  DCHECK_EQ(next_connect_state_, CONNECT_STATE_CONNECT_COMPLETE);
+
+  // Get the error that connect() completed with.
+  int os_error = 0;
+  socklen_t len = sizeof(os_error);
+  if (getsockopt(socket_, SOL_SOCKET, SO_ERROR, &os_error, &len) < 0)
+    os_error = errno;
+
+  // TODO(eroman): Is this check really necessary?
+  if (os_error == EINPROGRESS || os_error == EALREADY) {
+    NOTREACHED();  // This indicates a bug in libevent or our code.
+    return;
+  }
+
+  connect_os_error_ = os_error;
+  int rv = DoConnectLoop(MapConnectError(os_error));
+  if (rv != ERR_IO_PENDING) {
+    LogConnectCompletion(rv);
+    DoWriteCallback(rv);
+  }
+}
+
+void SCTPClientSocketLibevent::DidCompleteRead() {
+  int bytes_transferred;
+  bytes_transferred = HANDLE_EINTR(read(socket_, read_buf_->data(),
+                                        read_buf_len_));
+
+  int result;
+  if (bytes_transferred >= 0) {
+    result = bytes_transferred;
+    static base::StatsCounter read_bytes("sctp.read_bytes");
+    read_bytes.Add(bytes_transferred);
+    if (bytes_transferred > 0)
+      use_history_.set_was_used_to_convey_data();
+    LogByteTransfer(net_log_, NetLog::TYPE_SOCKET_BYTES_RECEIVED, result,
+                    read_buf_->data());
+  } else {
+    result = MapPosixError(errno);
+  }
+
+  if (result != ERR_IO_PENDING) {
+    read_buf_ = NULL;
+    read_buf_len_ = 0;
+    bool ok = read_socket_watcher_.StopWatchingFileDescriptor();
+    DCHECK(ok);
+    DoReadCallback(result);
+  }
+}
+
+void SCTPClientSocketLibevent::DidCompleteWrite() {
+  int bytes_transferred;
+  // TODO(jtl): Need to fix this to use sctp_sendmsg and need to get SCTP stream
+  // ID.  Perhaps use a form of IOBuffer that includes the SCTP stream ID?
+  bytes_transferred = HANDLE_EINTR(write(socket_, write_buf_->data(),

[Mike Belshe, 2011/04/06 18:32:53]

i think this should be InternalWrite()?

+                                         write_buf_len_));
+
+  int result;
+  if (bytes_transferred >= 0) {
+    result = bytes_transferred;
+    static base::StatsCounter write_bytes("sctp.write_bytes");
+    write_bytes.Add(bytes_transferred);
+    if (bytes_transferred > 0)
+      use_history_.set_was_used_to_convey_data();
+    LogByteTransfer(net_log_, NetLog::TYPE_SOCKET_BYTES_SENT, result,
+                    write_buf_->data());
+  } else {
+    result = MapPosixError(errno);
+  }
+
+  if (result != ERR_IO_PENDING) {
+    write_buf_ = NULL;
+    write_buf_len_ = 0;
+    write_socket_watcher_.StopWatchingFileDescriptor();
+    DoWriteCallback(result);
+  }
+}
+
+int SCTPClientSocketLibevent::GetPeerAddress(AddressList* address) const {
+  DCHECK(CalledOnValidThread());
+  DCHECK(address);
+  if (!IsConnected())
+    return ERR_SOCKET_NOT_CONNECTED;
+  address->Copy(current_ai_, false);
+  return OK;
+}
+
+const BoundNetLog& SCTPClientSocketLibevent::NetLog() const {
+  return net_log_;
+}
+
+}  // namespace net