Commit 2074c7fb authored by Guillaume Roguez's avatar Guillaume Roguez
Browse files

Ice: fix implementation

IceTransport::isXXX() API are not thread-safe, not clear in which
state the transport is and not easy to manipulate (order of state).

This patch tries to solve that by:
- Procted API by a mutex
- Remove uneeded isComplete() API
- Ensure that each API are ordered (one at true implies true
  on all lower states: INITIALIZED -> STARTED -> RUNNING).
  All return false on Fail state.
- Change usage accordingly

Tuleap: #107
Change-Id: I17211e54322d70bbfe18c28f06cf9967b9ef93d2
parent ef44e3dc
......@@ -102,7 +102,7 @@ IceTransport::IceTransport(const char* name, int component_count, bool master,
, on_negodone_cb_(options.onNegoDone)
, component_count_(component_count)
, compIO_(component_count)
, initiator_session_(master)
, initiatorSession_(master)
, thread_()
{
if (options.upnpEnable)
......@@ -248,40 +248,35 @@ IceTransport::onComplete(pj_ice_strans* ice_st, pj_ice_strans_op op,
pj_status_t status)
{
const char *opname =
op == PJ_ICE_STRANS_OP_INIT? "initialization" :
op == PJ_ICE_STRANS_OP_INIT ? "initialization" :
op == PJ_ICE_STRANS_OP_NEGOTIATION ? "negotiation" : "unknown_op";
if (!icest_.get())
icest_.reset(ice_st);
const bool done = status == PJ_SUCCESS;
RING_DBG("ICE %s with %s", opname, done?"success":"error");
if (!done)
RING_ERR("ICE %s failed: %s", opname,
sip_utils::sip_strerror(status).c_str());
if (done)
RING_DBG("ICE %s success", opname);
else
RING_ERR("ICE %s failed: %s", opname, sip_utils::sip_strerror(status).c_str());
{
std::lock_guard<std::mutex> lk(iceMutex_);
if (op == PJ_ICE_STRANS_OP_INIT) {
iceTransportInitDone_ = done;
if (iceTransportInitDone_) {
if (initiator_session_)
if (!icest_.get())
icest_.reset(ice_st);
}
if (done and op == PJ_ICE_STRANS_OP_INIT) {
if (initiatorSession_)
setInitiatorSession();
else
setSlaveSession();
selectUPnPIceCandidates();
}
} else if (op == PJ_ICE_STRANS_OP_NEGOTIATION) {
iceTransportNegoDone_ = done;
}
}
if (op == PJ_ICE_STRANS_OP_INIT and on_initdone_cb_)
on_initdone_cb_(*this, done);
else if (op == PJ_ICE_STRANS_OP_NEGOTIATION and on_negodone_cb_)
on_negodone_cb_(*this, done);
// Unlock waitForXXX APIs
iceCV_.notify_all();
}
......@@ -304,7 +299,7 @@ IceTransport::getDefaultCanditates()
bool
IceTransport::createIceSession(pj_ice_sess_role role)
{
if (pj_ice_strans_init_ice(icest_.get(), role, NULL, NULL) != PJ_SUCCESS) {
if (pj_ice_strans_init_ice(icest_.get(), role, nullptr, nullptr) != PJ_SUCCESS) {
RING_ERR("pj_ice_strans_init_ice() failed");
return false;
}
......@@ -320,7 +315,7 @@ bool
IceTransport::setInitiatorSession()
{
RING_DBG("ICE as master");
initiator_session_ = true;
initiatorSession_ = true;
if (isInitialized()) {
auto status = pj_ice_strans_change_role(icest_.get(), PJ_ICE_SESS_ROLE_CONTROLLING);
if (status != PJ_SUCCESS) {
......@@ -336,7 +331,7 @@ bool
IceTransport::setSlaveSession()
{
RING_DBG("ICE as slave");
initiator_session_ = false;
initiatorSession_ = false;
if (isInitialized()) {
auto status = pj_ice_strans_change_role(icest_.get(), PJ_ICE_SESS_ROLE_CONTROLLED);
if (status != PJ_SUCCESS) {
......@@ -353,7 +348,7 @@ IceTransport::isInitiator() const
{
if (isInitialized())
return pj_ice_strans_get_role(icest_.get()) == PJ_ICE_SESS_ROLE_CONTROLLING;
return initiator_session_;
return initiatorSession_;
}
bool
......@@ -441,50 +436,52 @@ IceTransport::start(const std::vector<uint8_t>& rem_data)
bool
IceTransport::stop()
{
if (not isInitialized()) {
RING_ERR("Session not created yet");
return false;
}
if (isStarted()) {
auto status = pj_ice_strans_stop_ice(icest_.get());
if (status != PJ_SUCCESS) {
RING_ERR("ICE start failed: %s", sip_utils::sip_strerror(status).c_str());
RING_ERR("ICE stop failed: %s", sip_utils::sip_strerror(status).c_str());
return false;
}
}
return true;
}
bool
IceTransport::isInitialized() const
IceTransport::_isInitialized() const
{
if (auto icest = icest_.get())
return pj_ice_strans_has_sess(icest);
if (auto icest = icest_.get()) {
auto state = pj_ice_strans_get_state(icest);
return state >= PJ_ICE_STRANS_STATE_SESS_READY and state != PJ_ICE_STRANS_STATE_FAILED;
}
return false;
}
bool
IceTransport::isStarted() const
IceTransport::_isStarted() const
{
return pj_ice_strans_sess_is_running(icest_.get());
}
bool
IceTransport::isCompleted() const
{
return pj_ice_strans_sess_is_complete(icest_.get());
if (auto icest = icest_.get()) {
auto state = pj_ice_strans_get_state(icest);
return state >= PJ_ICE_STRANS_STATE_NEGO and state != PJ_ICE_STRANS_STATE_FAILED;
}
return false;
}
bool
IceTransport::isRunning() const
IceTransport::_isRunning() const
{
return isInitialized() and pj_ice_strans_get_state(icest_.get()) == PJ_ICE_STRANS_STATE_RUNNING;
if (auto icest = icest_.get()) {
auto state = pj_ice_strans_get_state(icest);
return state >= PJ_ICE_STRANS_STATE_RUNNING and state != PJ_ICE_STRANS_STATE_FAILED;
}
return false;
}
bool
IceTransport::isFailed() const
IceTransport::_isFailed() const
{
return isInitialized() and pj_ice_strans_get_state(icest_.get()) == PJ_ICE_STRANS_STATE_FAILED;
if (auto icest = icest_.get())
return pj_ice_strans_get_state(icest) == PJ_ICE_STRANS_STATE_FAILED;
return false;
}
IpAddr
......@@ -560,6 +557,9 @@ IceTransport::getLocalCandidatesAddr(unsigned comp_id) const
bool
IceTransport::registerPublicIP(unsigned compId, const IpAddr& publicIP)
{
if (not isInitialized())
return false;
// Register only if no NAT traversal methods exists
if (upnp_ or config_.stun.port > 0 or config_.turn.port > 0)
return false;
......@@ -663,6 +663,9 @@ IceTransport::selectUPnPIceCandidates()
std::vector<uint8_t>
IceTransport::getLocalAttributesAndCandidates() const
{
if (not isInitialized())
return {};
std::stringstream ss;
ss << local_ufrag_ << NEW_LINE;
ss << local_pwd_ << NEW_LINE;
......@@ -672,8 +675,7 @@ IceTransport::getLocalAttributesAndCandidates() const
ss << c << NEW_LINE;
}
auto str(ss.str());
std::vector<uint8_t> ret(str.begin(), str.end());
return ret;
return std::vector<uint8_t>(str.begin(), str.end());
}
void
......@@ -786,11 +788,6 @@ IceTransport::setOnRecv(unsigned comp_id, IceRecvCb cb)
ssize_t
IceTransport::send(int comp_id, const unsigned char* buf, size_t len)
{
if (not isInitialized()) {
RING_ERR("ICE: not initialized transport");
return -1;
}
register_thread();
auto remote = getRemoteAddress(comp_id);
if (!remote) {
......@@ -828,12 +825,11 @@ IceTransport::waitForInitialization(unsigned timeout)
{
std::unique_lock<std::mutex> lk(iceMutex_);
if (!iceCV_.wait_for(lk, std::chrono::seconds(timeout),
[this]{ return iceTransportInitDone_; })) {
[this]{ return _isInitialized() or _isFailed(); })) {
RING_WARN("waitForInitialization: timeout");
return -1;
}
RING_DBG("waitForInitialization: %u", iceTransportInitDone_);
return iceTransportInitDone_;
return not _isFailed();
}
int
......@@ -841,12 +837,11 @@ IceTransport::waitForNegotiation(unsigned timeout)
{
std::unique_lock<std::mutex> lk(iceMutex_);
if (!iceCV_.wait_for(lk, std::chrono::seconds(timeout),
[this]{ return iceTransportNegoDone_; })) {
[this]{ return _isRunning() or _isFailed(); })) {
RING_WARN("waitForIceNegotiation: timeout");
return -1;
}
RING_DBG("waitForNegotiation: %u", iceTransportNegoDone_);
return iceTransportNegoDone_;
return not _isFailed();
}
ssize_t
......@@ -895,8 +890,7 @@ IceTransportFactory::createTransport(const char* name, int component_count,
const IceTransportOptions& options)
{
try {
return std::make_shared<IceTransport>(name, component_count, master,
options);
return std::make_shared<IceTransport>(name, component_count, master, options);
} catch(const std::exception& e) {
RING_ERR("%s",e.what());
return nullptr;
......
......@@ -112,14 +112,41 @@ class IceTransport {
*/
bool stop();
bool isInitialized() const;
/**
* Returns true if ICE transport has been initialized
* [mutex protected]
*/
bool isInitialized() const {
std::lock_guard<std::mutex> lk(iceMutex_);
return _isInitialized();
}
bool isStarted() const;
/**
* Returns true if ICE negotiation has been started
* [mutex protected]
*/
bool isStarted() const {
std::lock_guard<std::mutex> lk(iceMutex_);
return _isStarted();
}
bool isCompleted() const;
/**
* Returns true if ICE negotiation has completed with success
* [mutex protected]
*/
bool isRunning() const {
std::lock_guard<std::mutex> lk(iceMutex_);
return _isRunning();
}
bool isRunning() const;
bool isFailed() const;
/**
* Returns true if ICE transport is in failure state
* [mutex protected]
*/
bool isFailed() const {
std::lock_guard<std::mutex> lk(iceMutex_);
return _isFailed();
}
IpAddr getLocalAddress(unsigned comp_id) const;
......@@ -203,11 +230,15 @@ class IceTransport {
void getDefaultCanditates();
// Non-mutex protected of public versions
bool _isInitialized() const;
bool _isStarted() const;
bool _isRunning() const;
bool _isFailed() const;
std::unique_ptr<pj_pool_t, decltype(pj_pool_release)&> pool_;
IceTransportCompleteCb on_initdone_cb_;
IceTransportCompleteCb on_negodone_cb_;
bool iceTransportInitDone_ {false};
bool iceTransportNegoDone_ {false};
std::unique_ptr<pj_ice_strans, IceSTransDeleter> icest_;
unsigned component_count_;
pj_ice_sess_cand cand_[MAX_CANDIDATES] {};
......@@ -231,7 +262,7 @@ class IceTransport {
};
std::vector<ComponentIO> compIO_;
bool initiator_session_ {true};
std::atomic_bool initiatorSession_ {true};
/**
* Returns the IP of each candidate for a given component in the ICE session
......
......@@ -768,9 +768,9 @@ SIPCall::getAllRemoteCandidates()
bool
SIPCall::startIce()
{
if (not iceTransport_)
if (not iceTransport_ or iceTransport_->isFailed())
return false;
if (iceTransport_->isStarted() || iceTransport_->isCompleted()) {
if (iceTransport_->isStarted()) {
RING_DBG("[call:%s] ICE already started", getCallId().c_str());
return true;
}
......
Supports Markdown
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment