asio sync with upstream

This commit is contained in:
Marcos Pinto 2007-09-06 20:27:50 +00:00
commit 402bd70816
18 changed files with 237 additions and 141 deletions

View file

@ -238,6 +238,9 @@ public:
* with the asio::error::operation_aborted error. * with the asio::error::operation_aborted error.
* *
* @throws asio::system_error Thrown on failure. * @throws asio::system_error Thrown on failure.
*
* @note For portable behaviour with respect to graceful closure of a
* connected socket, call shutdown() before closing the socket.
*/ */
void close() void close()
{ {
@ -265,6 +268,9 @@ public:
* // An error occurred. * // An error occurred.
* } * }
* @endcode * @endcode
*
* @note For portable behaviour with respect to graceful closure of a
* connected socket, call shutdown() before closing the socket.
*/ */
asio::error_code close(asio::error_code& ec) asio::error_code close(asio::error_code& ec)
{ {

View file

@ -542,9 +542,10 @@ inline const_buffers_1 buffer(const PodType (&data)[N],
? N * sizeof(PodType) : max_size_in_bytes)); ? N * sizeof(PodType) : max_size_in_bytes));
} }
#if BOOST_WORKAROUND(__BORLANDC__, BOOST_TESTED_AT(0x582)) #if BOOST_WORKAROUND(__BORLANDC__, BOOST_TESTED_AT(0x582)) \
|| BOOST_WORKAROUND(__SUNPRO_CC, BOOST_TESTED_AT(0x590))
// Borland C++ thinks the overloads: // Borland C++ and Sun Studio think the overloads:
// //
// unspecified buffer(boost::array<PodType, N>& array ...); // unspecified buffer(boost::array<PodType, N>& array ...);
// //
@ -610,6 +611,7 @@ buffer(boost::array<PodType, N>& data, std::size_t max_size_in_bytes)
} }
#else // BOOST_WORKAROUND(__BORLANDC__, BOOST_TESTED_AT(0x582)) #else // BOOST_WORKAROUND(__BORLANDC__, BOOST_TESTED_AT(0x582))
// || BOOST_WORKAROUND(__SUNPRO_CC, BOOST_TESTED_AT(0x590))
/// Create a new modifiable buffer that represents the given POD array. /// Create a new modifiable buffer that represents the given POD array.
template <typename PodType, std::size_t N> template <typename PodType, std::size_t N>
@ -650,6 +652,7 @@ inline const_buffers_1 buffer(boost::array<const PodType, N>& data,
} }
#endif // BOOST_WORKAROUND(__BORLANDC__, BOOST_TESTED_AT(0x582)) #endif // BOOST_WORKAROUND(__BORLANDC__, BOOST_TESTED_AT(0x582))
// || BOOST_WORKAROUND(__SUNPRO_CC, BOOST_TESTED_AT(0x590))
/// Create a new non-modifiable buffer that represents the given POD array. /// Create a new non-modifiable buffer that represents the given POD array.
template <typename PodType, std::size_t N> template <typename PodType, std::size_t N>

View file

@ -43,17 +43,20 @@ public:
} }
// Signal the event. // Signal the event.
void signal() template <typename Lock>
void signal(Lock&)
{ {
} }
// Reset the event. // Reset the event.
void clear() template <typename Lock>
void clear(Lock&)
{ {
} }
// Wait for the event to become signalled. // Wait for the event to become signalled.
void wait() template <typename Lock>
void wait(Lock&)
{ {
} }
}; };

View file

@ -24,6 +24,7 @@
#if defined(BOOST_HAS_PTHREADS) #if defined(BOOST_HAS_PTHREADS)
#include "asio/detail/push_options.hpp" #include "asio/detail/push_options.hpp"
#include <boost/assert.hpp>
#include <boost/throw_exception.hpp> #include <boost/throw_exception.hpp>
#include <pthread.h> #include <pthread.h>
#include "asio/detail/pop_options.hpp" #include "asio/detail/pop_options.hpp"
@ -42,7 +43,7 @@ public:
posix_event() posix_event()
: signalled_(false) : signalled_(false)
{ {
int error = ::pthread_mutex_init(&mutex_, 0); int error = ::pthread_cond_init(&cond_, 0);
if (error != 0) if (error != 0)
{ {
asio::system_error e( asio::system_error e(
@ -50,53 +51,43 @@ public:
"event"); "event");
boost::throw_exception(e); boost::throw_exception(e);
} }
error = ::pthread_cond_init(&cond_, 0);
if (error != 0)
{
::pthread_mutex_destroy(&mutex_);
asio::system_error e(
asio::error_code(error, asio::native_ecat),
"event");
boost::throw_exception(e);
}
} }
// Destructor. // Destructor.
~posix_event() ~posix_event()
{ {
::pthread_cond_destroy(&cond_); ::pthread_cond_destroy(&cond_);
::pthread_mutex_destroy(&mutex_);
} }
// Signal the event. // Signal the event.
void signal() template <typename Lock>
void signal(Lock& lock)
{ {
::pthread_mutex_lock(&mutex_); // Ignore EINVAL and EDEADLK. BOOST_ASSERT(lock.locked());
(void)lock;
signalled_ = true; signalled_ = true;
::pthread_cond_signal(&cond_); // Ignore EINVAL. ::pthread_cond_signal(&cond_); // Ignore EINVAL.
::pthread_mutex_unlock(&mutex_); // Ignore EINVAL and EPERM.
} }
// Reset the event. // Reset the event.
void clear() template <typename Lock>
void clear(Lock& lock)
{ {
::pthread_mutex_lock(&mutex_); // Ignore EINVAL and EDEADLK. BOOST_ASSERT(lock.locked());
(void)lock;
signalled_ = false; signalled_ = false;
::pthread_mutex_unlock(&mutex_); // Ignore EINVAL and EPERM.
} }
// Wait for the event to become signalled. // Wait for the event to become signalled.
void wait() template <typename Lock>
void wait(Lock& lock)
{ {
::pthread_mutex_lock(&mutex_); // Ignore EINVAL and EDEADLK. BOOST_ASSERT(lock.locked());
while (!signalled_) while (!signalled_)
::pthread_cond_wait(&cond_, &mutex_); // Ignore EINVAL. ::pthread_cond_wait(&cond_, &lock.mutex().mutex_); // Ignore EINVAL.
::pthread_mutex_unlock(&mutex_); // Ignore EINVAL and EPERM.
} }
private: private:
::pthread_mutex_t mutex_;
::pthread_cond_t cond_; ::pthread_cond_t cond_;
bool signalled_; bool signalled_;
}; };

View file

@ -35,6 +35,8 @@
namespace asio { namespace asio {
namespace detail { namespace detail {
class posix_event;
class posix_mutex class posix_mutex
: private noncopyable : private noncopyable
{ {
@ -87,6 +89,7 @@ public:
} }
private: private:
friend class posix_event;
::pthread_mutex_t mutex_; ::pthread_mutex_t mutex_;
}; };

View file

@ -86,7 +86,7 @@ public:
}; };
// The maximum number of buffers to support in a single operation. // The maximum number of buffers to support in a single operation.
enum { max_buffers = 16 }; enum { max_buffers = 64 < max_iov_len ? 64 : max_iov_len };
// Constructor. // Constructor.
reactive_socket_service(asio::io_service& io_service) reactive_socket_service(asio::io_service& io_service)

View file

@ -63,6 +63,18 @@ public:
} }
} }
// Test whether the lock is held.
bool locked() const
{
return locked_;
}
// Get the underlying mutex.
Mutex& mutex()
{
return mutex_;
}
private: private:
// The underlying mutex. // The underlying mutex.
Mutex& mutex_; Mutex& mutex_;

View file

@ -166,7 +166,8 @@ private:
} }
// Check if a service matches the given id. // Check if a service matches the given id.
bool service_id_matches(const asio::io_service::service& service, static bool service_id_matches(
const asio::io_service::service& service,
const asio::io_service::id& id) const asio::io_service::id& id)
{ {
return service.id_ == &id; return service.id_ == &id;
@ -174,7 +175,8 @@ private:
// Check if a service matches the given id. // Check if a service matches the given id.
template <typename Service> template <typename Service>
bool service_id_matches(const asio::io_service::service& service, static bool service_id_matches(
const asio::io_service::service& service,
const asio::detail::service_id<Service>& /*id*/) const asio::detail::service_id<Service>& /*id*/)
{ {
return service.type_info_ != 0 && *service.type_info_ == typeid(Service); return service.type_info_ != 0 && *service.type_info_ == typeid(Service);

View file

@ -110,9 +110,20 @@ public:
template <typename Protocol> template <typename Protocol>
void resize(const Protocol&, std::size_t s) void resize(const Protocol&, std::size_t s)
{ {
if (s != sizeof(value_)) // On some platforms (e.g. Windows Vista), the getsockopt function will
// return the size of a boolean socket option as one byte, even though a
// four byte integer was passed in.
switch (s)
{
case sizeof(char):
value_ = *reinterpret_cast<char*>(&value_) ? 1 : 0;
break;
case sizeof(value_):
break;
default:
throw std::length_error("boolean socket option resize"); throw std::length_error("boolean socket option resize");
} }
}
private: private:
int value_; int value_;

View file

@ -98,6 +98,7 @@
# include <arpa/inet.h> # include <arpa/inet.h>
# include <netdb.h> # include <netdb.h>
# include <net/if.h> # include <net/if.h>
# include <limits.h>
# if defined(__sun) # if defined(__sun)
# include <sys/filio.h> # include <sys/filio.h>
# include <sys/sockio.h> # include <sys/sockio.h>
@ -141,6 +142,11 @@ const int shutdown_both = SD_BOTH;
const int message_peek = MSG_PEEK; const int message_peek = MSG_PEEK;
const int message_out_of_band = MSG_OOB; const int message_out_of_band = MSG_OOB;
const int message_do_not_route = MSG_DONTROUTE; const int message_do_not_route = MSG_DONTROUTE;
# if defined (_WIN32_WINNT)
const int max_iov_len = 64;
# else
const int max_iov_len = 16;
# endif
#else #else
typedef int socket_type; typedef int socket_type;
const int invalid_socket = -1; const int invalid_socket = -1;
@ -166,6 +172,7 @@ const int shutdown_both = SHUT_RDWR;
const int message_peek = MSG_PEEK; const int message_peek = MSG_PEEK;
const int message_out_of_band = MSG_OOB; const int message_out_of_band = MSG_OOB;
const int message_do_not_route = MSG_DONTROUTE; const int message_do_not_route = MSG_DONTROUTE;
const int max_iov_len = IOV_MAX;
#endif #endif
const int custom_socket_option_level = 0xA5100000; const int custom_socket_option_level = 0xA5100000;
const int enable_connection_aborted_option = 1; const int enable_connection_aborted_option = 1;

View file

@ -239,6 +239,7 @@ public:
#else #else
BOOST_ASSERT(size <= strand_impl::handler_storage_type::size); BOOST_ASSERT(size <= strand_impl::handler_storage_type::size);
#endif #endif
(void)size;
return impl_->handler_storage_.address(); return impl_->handler_storage_.address();
} }
@ -415,14 +416,14 @@ public:
} }
else else
{ {
asio::detail::mutex::scoped_lock lock(impl->mutex_);
// Allocate and construct an object to wrap the handler. // Allocate and construct an object to wrap the handler.
typedef handler_wrapper<Handler> value_type; typedef handler_wrapper<Handler> value_type;
typedef handler_alloc_traits<Handler, value_type> alloc_traits; typedef handler_alloc_traits<Handler, value_type> alloc_traits;
raw_handler_ptr<alloc_traits> raw_ptr(handler); raw_handler_ptr<alloc_traits> raw_ptr(handler);
handler_ptr<alloc_traits> ptr(raw_ptr, handler); handler_ptr<alloc_traits> ptr(raw_ptr, handler);
asio::detail::mutex::scoped_lock lock(impl->mutex_);
if (impl->current_handler_ == 0) if (impl->current_handler_ == 0)
{ {
// This handler now has the lock, so can be dispatched immediately. // This handler now has the lock, so can be dispatched immediately.
@ -455,14 +456,14 @@ public:
template <typename Handler> template <typename Handler>
void post(implementation_type& impl, Handler handler) void post(implementation_type& impl, Handler handler)
{ {
asio::detail::mutex::scoped_lock lock(impl->mutex_);
// Allocate and construct an object to wrap the handler. // Allocate and construct an object to wrap the handler.
typedef handler_wrapper<Handler> value_type; typedef handler_wrapper<Handler> value_type;
typedef handler_alloc_traits<Handler, value_type> alloc_traits; typedef handler_alloc_traits<Handler, value_type> alloc_traits;
raw_handler_ptr<alloc_traits> raw_ptr(handler); raw_handler_ptr<alloc_traits> raw_ptr(handler);
handler_ptr<alloc_traits> ptr(raw_ptr, handler); handler_ptr<alloc_traits> ptr(raw_ptr, handler);
asio::detail::mutex::scoped_lock lock(impl->mutex_);
if (impl->current_handler_ == 0) if (impl->current_handler_ == 0)
{ {
// This handler now has the lock, so can be dispatched immediately. // This handler now has the lock, so can be dispatched immediately.

View file

@ -40,6 +40,7 @@ public:
: asio::detail::service_base<task_io_service<Task> >(io_service), : asio::detail::service_base<task_io_service<Task> >(io_service),
mutex_(), mutex_(),
task_(use_service<Task>(io_service)), task_(use_service<Task>(io_service)),
task_interrupted_(true),
outstanding_work_(0), outstanding_work_(0),
handler_queue_(&task_handler_), handler_queue_(&task_handler_),
handler_queue_end_(&task_handler_), handler_queue_end_(&task_handler_),
@ -80,8 +81,7 @@ public:
typename call_stack<task_io_service>::context ctx(this); typename call_stack<task_io_service>::context ctx(this);
idle_thread_info this_idle_thread; idle_thread_info this_idle_thread;
this_idle_thread.prev = &this_idle_thread; this_idle_thread.next = 0;
this_idle_thread.next = &this_idle_thread;
asio::detail::mutex::scoped_lock lock(mutex_); asio::detail::mutex::scoped_lock lock(mutex_);
@ -98,8 +98,7 @@ public:
typename call_stack<task_io_service>::context ctx(this); typename call_stack<task_io_service>::context ctx(this);
idle_thread_info this_idle_thread; idle_thread_info this_idle_thread;
this_idle_thread.prev = &this_idle_thread; this_idle_thread.next = 0;
this_idle_thread.next = &this_idle_thread;
asio::detail::mutex::scoped_lock lock(mutex_); asio::detail::mutex::scoped_lock lock(mutex_);
@ -134,7 +133,7 @@ public:
void stop() void stop()
{ {
asio::detail::mutex::scoped_lock lock(mutex_); asio::detail::mutex::scoped_lock lock(mutex_);
stop_all_threads(); stop_all_threads(lock);
} }
// Reset in preparation for a subsequent run invocation. // Reset in preparation for a subsequent run invocation.
@ -156,7 +155,7 @@ public:
{ {
asio::detail::mutex::scoped_lock lock(mutex_); asio::detail::mutex::scoped_lock lock(mutex_);
if (--outstanding_work_ == 0) if (--outstanding_work_ == 0)
stop_all_threads(); stop_all_threads(lock);
} }
// Request invocation of the given handler. // Request invocation of the given handler.
@ -201,10 +200,15 @@ public:
++outstanding_work_; ++outstanding_work_;
// Wake up a thread to execute the handler. // Wake up a thread to execute the handler.
if (!interrupt_one_idle_thread()) if (!interrupt_one_idle_thread(lock))
if (task_handler_.next_ == 0 && handler_queue_end_ != &task_handler_) {
if (!task_interrupted_)
{
task_interrupted_ = true;
task_.interrupt(); task_.interrupt();
} }
}
}
private: private:
struct idle_thread_info; struct idle_thread_info;
@ -214,7 +218,7 @@ private:
{ {
if (outstanding_work_ == 0 && !stopped_) if (outstanding_work_ == 0 && !stopped_)
{ {
stop_all_threads(); stop_all_threads(lock);
ec = asio::error_code(); ec = asio::error_code();
return 0; return 0;
} }
@ -230,11 +234,14 @@ private:
handler_queue_ = h->next_; handler_queue_ = h->next_;
if (handler_queue_ == 0) if (handler_queue_ == 0)
handler_queue_end_ = 0; handler_queue_end_ = 0;
bool more_handlers = (handler_queue_ != 0); h->next_ = 0;
lock.unlock();
if (h == &task_handler_) if (h == &task_handler_)
{ {
bool more_handlers = (handler_queue_ != 0);
task_interrupted_ = more_handlers || polling;
lock.unlock();
// If the task has already run and we're polling then we're done. // If the task has already run and we're polling then we're done.
if (task_has_run && polling) if (task_has_run && polling)
{ {
@ -252,6 +259,7 @@ private:
} }
else else
{ {
lock.unlock();
handler_cleanup c(lock, *this); handler_cleanup c(lock, *this);
// Invoke the handler. May throw an exception. // Invoke the handler. May throw an exception.
@ -264,31 +272,10 @@ private:
else if (this_idle_thread) else if (this_idle_thread)
{ {
// Nothing to run right now, so just wait for work to do. // Nothing to run right now, so just wait for work to do.
if (first_idle_thread_)
{
this_idle_thread->next = first_idle_thread_; this_idle_thread->next = first_idle_thread_;
this_idle_thread->prev = first_idle_thread_->prev;
first_idle_thread_->prev->next = this_idle_thread;
first_idle_thread_->prev = this_idle_thread;
}
first_idle_thread_ = this_idle_thread; first_idle_thread_ = this_idle_thread;
this_idle_thread->wakeup_event.clear(); this_idle_thread->wakeup_event.clear(lock);
lock.unlock(); this_idle_thread->wakeup_event.wait(lock);
this_idle_thread->wakeup_event.wait();
lock.lock();
if (this_idle_thread->next == this_idle_thread)
{
first_idle_thread_ = 0;
}
else
{
if (first_idle_thread_ == this_idle_thread)
first_idle_thread_ = this_idle_thread->next;
this_idle_thread->next->prev = this_idle_thread->prev;
this_idle_thread->prev->next = this_idle_thread->next;
this_idle_thread->next = this_idle_thread;
this_idle_thread->prev = this_idle_thread;
}
} }
else else
{ {
@ -302,39 +289,44 @@ private:
} }
// Stop the task and all idle threads. // Stop the task and all idle threads.
void stop_all_threads() void stop_all_threads(
asio::detail::mutex::scoped_lock& lock)
{ {
stopped_ = true; stopped_ = true;
interrupt_all_idle_threads(); interrupt_all_idle_threads(lock);
if (task_handler_.next_ == 0 && handler_queue_end_ != &task_handler_) if (!task_interrupted_)
{
task_interrupted_ = true;
task_.interrupt(); task_.interrupt();
} }
}
// Interrupt a single idle thread. Returns true if a thread was interrupted, // Interrupt a single idle thread. Returns true if a thread was interrupted,
// false if no running thread could be found to interrupt. // false if no running thread could be found to interrupt.
bool interrupt_one_idle_thread() bool interrupt_one_idle_thread(
asio::detail::mutex::scoped_lock& lock)
{ {
if (first_idle_thread_) if (first_idle_thread_)
{ {
first_idle_thread_->wakeup_event.signal(); idle_thread_info* idle_thread = first_idle_thread_;
first_idle_thread_ = first_idle_thread_->next; first_idle_thread_ = idle_thread->next;
idle_thread->next = 0;
idle_thread->wakeup_event.signal(lock);
return true; return true;
} }
return false; return false;
} }
// Interrupt all idle threads. // Interrupt all idle threads.
void interrupt_all_idle_threads() void interrupt_all_idle_threads(
asio::detail::mutex::scoped_lock& lock)
{ {
if (first_idle_thread_) while (first_idle_thread_)
{ {
first_idle_thread_->wakeup_event.signal(); idle_thread_info* idle_thread = first_idle_thread_;
idle_thread_info* current_idle_thread = first_idle_thread_->next; first_idle_thread_ = idle_thread->next;
while (current_idle_thread != first_idle_thread_) idle_thread->next = 0;
{ idle_thread->wakeup_event.signal(lock);
current_idle_thread->wakeup_event.signal();
current_idle_thread = current_idle_thread->next;
}
} }
} }
@ -440,6 +432,7 @@ private:
{ {
// Reinsert the task at the end of the handler queue. // Reinsert the task at the end of the handler queue.
lock_.lock(); lock_.lock();
task_io_service_.task_interrupted_ = true;
task_io_service_.task_handler_.next_ = 0; task_io_service_.task_handler_.next_ = 0;
if (task_io_service_.handler_queue_end_) if (task_io_service_.handler_queue_end_)
{ {
@ -478,7 +471,7 @@ private:
{ {
lock_.lock(); lock_.lock();
if (--task_io_service_.outstanding_work_ == 0) if (--task_io_service_.outstanding_work_ == 0)
task_io_service_.stop_all_threads(); task_io_service_.stop_all_threads(lock_);
} }
private: private:
@ -503,6 +496,9 @@ private:
} }
} task_handler_; } task_handler_;
// Whether the task has been interrupted.
bool task_interrupted_;
// The count of unfinished work. // The count of unfinished work.
int outstanding_work_; int outstanding_work_;
@ -522,7 +518,6 @@ private:
struct idle_thread_info struct idle_thread_info
{ {
event wakeup_event; event wakeup_event;
idle_thread_info* prev;
idle_thread_info* next; idle_thread_info* next;
}; };

View file

@ -28,6 +28,7 @@
#include "asio/detail/socket_types.hpp" #include "asio/detail/socket_types.hpp"
#include "asio/detail/push_options.hpp" #include "asio/detail/push_options.hpp"
#include <boost/assert.hpp>
#include <boost/throw_exception.hpp> #include <boost/throw_exception.hpp>
#include "asio/detail/pop_options.hpp" #include "asio/detail/pop_options.hpp"
@ -59,21 +60,31 @@ public:
} }
// Signal the event. // Signal the event.
void signal() template <typename Lock>
void signal(Lock& lock)
{ {
BOOST_ASSERT(lock.locked());
(void)lock;
::SetEvent(event_); ::SetEvent(event_);
} }
// Reset the event. // Reset the event.
void clear() template <typename Lock>
void clear(Lock& lock)
{ {
BOOST_ASSERT(lock.locked());
(void)lock;
::ResetEvent(event_); ::ResetEvent(event_);
} }
// Wait for the event to become signalled. // Wait for the event to become signalled.
void wait() template <typename Lock>
void wait(Lock& lock)
{ {
BOOST_ASSERT(lock.locked());
lock.unlock();
::WaitForSingleObject(event_, INFINITE); ::WaitForSingleObject(event_, INFINITE);
lock.lock();
} }
private: private:

View file

@ -137,7 +137,7 @@ public:
enum enum
{ {
enable_connection_aborted = 1, // User wants connection_aborted errors. enable_connection_aborted = 1, // User wants connection_aborted errors.
user_set_linger = 2, // The user set the linger option. close_might_block = 2, // User set linger option for blocking close.
user_set_non_blocking = 4 // The user wants a non-blocking socket. user_set_non_blocking = 4 // The user wants a non-blocking socket.
}; };
@ -170,7 +170,7 @@ public:
typedef detail::select_reactor<true> reactor_type; typedef detail::select_reactor<true> reactor_type;
// The maximum number of buffers to support in a single operation. // The maximum number of buffers to support in a single operation.
enum { max_buffers = 16 }; enum { max_buffers = 64 < max_iov_len ? 64 : max_iov_len };
// Constructor. // Constructor.
win_iocp_socket_service(asio::io_service& io_service) win_iocp_socket_service(asio::io_service& io_service)
@ -192,7 +192,7 @@ public:
while (impl) while (impl)
{ {
asio::error_code ignored_ec; asio::error_code ignored_ec;
close(*impl, ignored_ec); close_for_destruction(*impl);
impl = impl->next_; impl = impl->next_;
} }
} }
@ -217,34 +217,7 @@ public:
// Destroy a socket implementation. // Destroy a socket implementation.
void destroy(implementation_type& impl) void destroy(implementation_type& impl)
{ {
if (impl.socket_ != invalid_socket) close_for_destruction(impl);
{
// Check if the reactor was created, in which case we need to close the
// socket on the reactor as well to cancel any operations that might be
// running there.
reactor_type* reactor = static_cast<reactor_type*>(
interlocked_compare_exchange_pointer(
reinterpret_cast<void**>(&reactor_), 0, 0));
if (reactor)
reactor->close_descriptor(impl.socket_);
if (impl.flags_ & implementation_type::user_set_linger)
{
::linger opt;
opt.l_onoff = 0;
opt.l_linger = 0;
asio::error_code ignored_ec;
socket_ops::setsockopt(impl.socket_,
SOL_SOCKET, SO_LINGER, &opt, sizeof(opt), ignored_ec);
}
asio::error_code ignored_ec;
socket_ops::close(impl.socket_, ignored_ec);
impl.socket_ = invalid_socket;
impl.flags_ = 0;
impl.cancel_token_.reset();
impl.safe_cancellation_thread_id_ = 0;
}
// Remove implementation from linked list of all implementations. // Remove implementation from linked list of all implementations.
asio::detail::mutex::scoped_lock lock(mutex_); asio::detail::mutex::scoped_lock lock(mutex_);
@ -353,6 +326,24 @@ public:
{ {
ec = asio::error::bad_descriptor; ec = asio::error::bad_descriptor;
} }
else if (FARPROC cancel_io_ex_ptr = ::GetProcAddress(
::GetModuleHandle("KERNEL32"), "CancelIoEx"))
{
// The version of Windows supports cancellation from any thread.
typedef BOOL (WINAPI* cancel_io_ex_t)(HANDLE, LPOVERLAPPED);
cancel_io_ex_t cancel_io_ex = (cancel_io_ex_t)cancel_io_ex_ptr;
socket_type sock = impl.socket_;
HANDLE sock_as_handle = reinterpret_cast<HANDLE>(sock);
if (!cancel_io_ex(sock_as_handle, 0))
{
DWORD last_error = ::GetLastError();
ec = asio::error_code(last_error, asio::native_ecat);
}
else
{
ec = asio::error_code();
}
}
else if (impl.safe_cancellation_thread_id_ == 0) else if (impl.safe_cancellation_thread_id_ == 0)
{ {
// No operations have been started, so there's nothing to cancel. // No operations have been started, so there's nothing to cancel.
@ -475,7 +466,12 @@ public:
if (option.level(impl.protocol_) == SOL_SOCKET if (option.level(impl.protocol_) == SOL_SOCKET
&& option.name(impl.protocol_) == SO_LINGER) && option.name(impl.protocol_) == SO_LINGER)
{ {
impl.flags_ |= implementation_type::user_set_linger; const ::linger* linger_option =
reinterpret_cast<const ::linger*>(option.data(impl.protocol_));
if (linger_option->l_onoff != 0 && linger_option->l_linger != 0)
impl.flags_ |= implementation_type::close_might_block;
else
impl.flags_ &= ~implementation_type::close_might_block;
} }
socket_ops::setsockopt(impl.socket_, socket_ops::setsockopt(impl.socket_,
@ -1950,6 +1946,43 @@ public:
} }
private: private:
// Helper function to close a socket when the associated object is being
// destroyed.
void close_for_destruction(implementation_type& impl)
{
if (is_open(impl))
{
// Check if the reactor was created, in which case we need to close the
// socket on the reactor as well to cancel any operations that might be
// running there.
reactor_type* reactor = static_cast<reactor_type*>(
interlocked_compare_exchange_pointer(
reinterpret_cast<void**>(&reactor_), 0, 0));
if (reactor)
reactor->close_descriptor(impl.socket_);
// The socket destructor must not block. If the user has changed the
// linger option to block in the foreground, we will change it back to the
// default so that the closure is performed in the background.
if (impl.flags_ & implementation_type::close_might_block)
{
::linger opt;
opt.l_onoff = 0;
opt.l_linger = 0;
asio::error_code ignored_ec;
socket_ops::setsockopt(impl.socket_,
SOL_SOCKET, SO_LINGER, &opt, sizeof(opt), ignored_ec);
}
asio::error_code ignored_ec;
socket_ops::close(impl.socket_, ignored_ec);
impl.socket_ = invalid_socket;
impl.flags_ = 0;
impl.cancel_token_.reset();
impl.safe_cancellation_thread_id_ = 0;
}
}
// Helper function to emulate InterlockedCompareExchangePointer functionality // Helper function to emulate InterlockedCompareExchangePointer functionality
// for: // for:
// - very old Platform SDKs; and // - very old Platform SDKs; and

View file

@ -17,6 +17,10 @@
#include "asio/detail/push_options.hpp" #include "asio/detail/push_options.hpp"
#include "asio/detail/push_options.hpp"
#include <boost/type_traits.hpp>
#include "asio/detail/pop_options.hpp"
#include "asio/detail/bind_handler.hpp" #include "asio/detail/bind_handler.hpp"
#include "asio/detail/handler_alloc_helpers.hpp" #include "asio/detail/handler_alloc_helpers.hpp"
#include "asio/detail/handler_invoke_helpers.hpp" #include "asio/detail/handler_invoke_helpers.hpp"
@ -30,7 +34,9 @@ class wrapped_handler
public: public:
typedef void result_type; typedef void result_type;
wrapped_handler(Dispatcher& dispatcher, Handler handler) wrapped_handler(
typename boost::add_reference<Dispatcher>::type dispatcher,
Handler handler)
: dispatcher_(dispatcher), : dispatcher_(dispatcher),
handler_(handler) handler_(handler)
{ {
@ -117,7 +123,7 @@ public:
} }
//private: //private:
Dispatcher& dispatcher_; Dispatcher dispatcher_;
Handler handler_; Handler handler_;
}; };
@ -171,9 +177,9 @@ inline void asio_handler_invoke(const Function& function,
function, this_handler->handler_)); function, this_handler->handler_));
} }
template <typename Function, typename Dispatcher, typename Handler> template <typename Function, typename Handler, typename Context>
inline void asio_handler_invoke(const Function& function, inline void asio_handler_invoke(const Function& function,
rewrapped_handler<Dispatcher, Handler>* this_handler) rewrapped_handler<Handler, Context>* this_handler)
{ {
asio_handler_invoke_helpers::invoke( asio_handler_invoke_helpers::invoke(
function, &this_handler->context_); function, &this_handler->context_);

View file

@ -128,11 +128,11 @@ template <typename Handler>
#if defined(GENERATING_DOCUMENTATION) #if defined(GENERATING_DOCUMENTATION)
unspecified unspecified
#else #else
inline detail::wrapped_handler<io_service, Handler> inline detail::wrapped_handler<io_service&, Handler>
#endif #endif
io_service::wrap(Handler handler) io_service::wrap(Handler handler)
{ {
return detail::wrapped_handler<io_service, Handler>(*this, handler); return detail::wrapped_handler<io_service&, Handler>(*this, handler);
} }
inline io_service::work::work(asio::io_service& io_service) inline io_service::work::work(asio::io_service& io_service)

View file

@ -320,7 +320,7 @@ public:
#if defined(GENERATING_DOCUMENTATION) #if defined(GENERATING_DOCUMENTATION)
unspecified unspecified
#else #else
detail::wrapped_handler<io_service, Handler> detail::wrapped_handler<io_service&, Handler>
#endif #endif
wrap(Handler handler); wrap(Handler handler);

View file

@ -172,7 +172,7 @@ public:
/// The protocol associated with the endpoint. /// The protocol associated with the endpoint.
protocol_type protocol() const protocol_type protocol() const
{ {
if (is_v4()) if (is_v4(data_))
return InternetProtocol::v4(); return InternetProtocol::v4();
return InternetProtocol::v6(); return InternetProtocol::v6();
} }
@ -192,7 +192,7 @@ public:
/// Get the underlying size of the endpoint in the native type. /// Get the underlying size of the endpoint in the native type.
size_type size() const size_type size() const
{ {
if (is_v4()) if (is_v4(data_))
return sizeof(asio::detail::sockaddr_in4_type); return sizeof(asio::detail::sockaddr_in4_type);
else else
return sizeof(asio::detail::sockaddr_in6_type); return sizeof(asio::detail::sockaddr_in6_type);
@ -218,7 +218,7 @@ public:
/// the host's byte order. /// the host's byte order.
unsigned short port() const unsigned short port() const
{ {
if (is_v4()) if (is_v4(data_))
{ {
return asio::detail::socket_ops::network_to_host_short( return asio::detail::socket_ops::network_to_host_short(
reinterpret_cast<const asio::detail::sockaddr_in4_type&>( reinterpret_cast<const asio::detail::sockaddr_in4_type&>(
@ -236,7 +236,7 @@ public:
/// the host's byte order. /// the host's byte order.
void port(unsigned short port_num) void port(unsigned short port_num)
{ {
if (is_v4()) if (is_v4(data_))
{ {
reinterpret_cast<asio::detail::sockaddr_in4_type&>(data_).sin_port reinterpret_cast<asio::detail::sockaddr_in4_type&>(data_).sin_port
= asio::detail::socket_ops::host_to_network_short(port_num); = asio::detail::socket_ops::host_to_network_short(port_num);
@ -252,7 +252,7 @@ public:
asio::ip::address address() const asio::ip::address address() const
{ {
using namespace std; // For memcpy. using namespace std; // For memcpy.
if (is_v4()) if (is_v4(data_))
{ {
const asio::detail::sockaddr_in4_type& data const asio::detail::sockaddr_in4_type& data
= reinterpret_cast<const asio::detail::sockaddr_in4_type&>( = reinterpret_cast<const asio::detail::sockaddr_in4_type&>(
@ -306,15 +306,27 @@ public:
private: private:
// Helper function to determine whether the endpoint is IPv4. // Helper function to determine whether the endpoint is IPv4.
bool is_v4() const
{
#if defined(_AIX) #if defined(_AIX)
return data_.__ss_family == AF_INET; template <typename T, unsigned short (T::*)> struct is_v4_helper {};
#else
return data_.ss_family == AF_INET; template <typename T>
#endif static bool is_v4(const T& ss, is_v4_helper<T, &T::ss_family>* = 0)
{
return ss.ss_family == AF_INET;
} }
template <typename T>
static bool is_v4(const T& ss, is_v4_helper<T, &T::__ss_family>* = 0)
{
return ss.__ss_family == AF_INET;
}
#else
static bool is_v4(const asio::detail::sockaddr_storage_type& ss)
{
return ss.ss_family == AF_INET;
}
#endif
// The underlying IP socket address. // The underlying IP socket address.
asio::detail::sockaddr_storage_type data_; asio::detail::sockaddr_storage_type data_;
}; };