<commit_before>/**
 * @file posix/net.cpp
 * @brief POSIX network access layer (using cURL + c-ares)
 *
 * (c) 2013-2014 by Mega Limited, Auckland, New Zealand
 *
 * This file is part of the MEGA SDK - Client Access Engine.
 *
 * Applications using the MEGA API must present a valid application key
 * and comply with the the rules set forth in the Terms of Service.
 *
 * The MEGA SDK is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
 *
 * @copyright Simplified (2-clause) BSD License.
 *
 * You should have received a copy of the license along with this
 * program.
 */

#include "mega.h"
#include "mega/posix/meganet.h"
#include "mega/logging.h"

#define IPV6_RETRY_INTERVAL_DS 72000
#define DNS_CACHE_TIMEOUT_DS 18000
#define MAX_SPEED_CONTROL_TIMEOUT_MS 500

namespace mega {

MUTEX_CLASS CurlHttpIO::curlMutex(false);

#if !defined(USE_CURL_PUBLIC_KEY_PINNING) || defined(WINDOWS_PHONE)

MUTEX_CLASS **CurlHttpIO::sslMutexes = NULL;
void CurlHttpIO::locking_function(int mode, int lockNumber, const char *, int)
{
    MUTEX_CLASS *mutex = sslMutexes[lockNumber];
    if (mutex == NULL)
    {
        mutex = new MUTEX_CLASS(true);
        sslMutexes[lockNumber] = mutex;
    }

    if (mode & CRYPTO_LOCK)
    {
        mutex->lock();
    }
    else
    {
        mutex->unlock();
    }
}

#if OPENSSL_VERSION_NUMBER >= 0x10000000
void CurlHttpIO::id_function(CRYPTO_THREADID* id)
{
    CRYPTO_THREADID_set_pointer(id, (void *)THREAD_CLASS::currentThreadId());
}
#else
unsigned long CurlHttpIO::id_function()
{
    return THREAD_CLASS::currentThreadId();
}
#endif

#endif

CurlHttpIO::CurlHttpIO()
{
    curl_version_info_data* data = curl_version_info(CURLVERSION_NOW);
    string curlssl = data->ssl_version;
    std::transform(curlssl.begin(), curlssl.end(), curlssl.begin(), ::tolower);

#if !defined(USE_CURL_PUBLIC_KEY_PINNING) || defined(WINDOWS_PHONE)
    if (!strstr(curlssl.c_str(), "openssl"))
    {
        LOG_fatal << "cURL built without OpenSSL support. Aborting.";
        exit(EXIT_FAILURE);
    }
#endif

    int i;

    for (i = 0; data->protocols[i]; i++)
    {
        if (strstr(data->protocols[i], "http"))
        {
            break;
        }
    }

    if (!data->protocols[i] || !(data->features & CURL_VERSION_SSL))
    {
        LOG_fatal << "cURL built without HTTP/HTTPS support. Aborting.";
        exit(EXIT_FAILURE);
    }

    curlipv6 = data->features & CURL_VERSION_IPV6;
    LOG_debug << "IPv6 enabled: " << curlipv6;

    dnsok = false;
    reset = false;
    statechange = false;
    maxdownloadspeed = 0;
    maxuploadspeed = 0;

    WAIT_CLASS::bumpds();
    lastdnspurge = Waiter::ds + DNS_CACHE_TIMEOUT_DS / 2;

    curlMutex.lock();

#if !defined(USE_CURL_PUBLIC_KEY_PINNING) || defined(WINDOWS_PHONE)

    if (!CRYPTO_get_locking_callback()
#if OPENSSL_VERSION_NUMBER >= 0x10000000
        && !CRYPTO_THREADID_get_callback())
#else
        && !CRYPTO_get_id_callback())
#endif
    {
        LOG_debug << "Initializing OpenSSL locking callbacks";
        int numLocks = CRYPTO_num_locks();
        sslMutexes = new MUTEX_CLASS*[numLocks];
        memset(sslMutexes, 0, numLocks * sizeof(MUTEX_CLASS*));
#if OPENSSL_VERSION_NUMBER >= 0x10000000
        CRYPTO_THREADID_set_callback(CurlHttpIO::id_function);
#else
        CRYPTO_set_id_callback(CurlHttpIO::id_function);
#endif
        CRYPTO_set_locking_callback(CurlHttpIO::locking_function);
    }

#endif

    curl_global_init(CURL_GLOBAL_DEFAULT);
    ares_library_init(ARES_LIB_INIT_ALL);
    curlMutex.unlock();

    curlmapi = curl_multi_init();
    curlmdownload = curl_multi_init();
    curlmupload = curl_multi_init();

    struct ares_options options;
    options.tries = 2;
    ares_init_options(&ares, &options, ARES_OPT_TRIES);
    filterDNSservers();

#if defined(_WIN32) && !defined(WINDOWS_PHONE)
    curl_multi_setopt(curlmapi, CURLMOPT_SOCKETFUNCTION, api_socket_callback);
    curl_multi_setopt(curlmapi, CURLMOPT_SOCKETDATA, this);
    curl_multi_setopt(curlmapi, CURLMOPT_TIMERFUNCTION, api_timer_callback);
    curl_multi_setopt(curlmapi, CURLMOPT_TIMERDATA, this);
    curlapitimeoutreset = 0;

    curl_multi_setopt(curlmdownload, CURLMOPT_SOCKETFUNCTION, download_socket_callback);
    curl_multi_setopt(curlmdownload, CURLMOPT_SOCKETDATA, this);
    curl_multi_setopt(curlmdownload, CURLMOPT_TIMERFUNCTION, download_timer_callback);
    curl_multi_setopt(curlmdownload, CURLMOPT_TIMERDATA, this);
    curldownloadtimeoutreset = 0;

    curl_multi_setopt(curlmupload, CURLMOPT_SOCKETFUNCTION, upload_socket_callback);
    curl_multi_setopt(curlmupload, CURLMOPT_SOCKETDATA, this);
    curl_multi_setopt(curlmupload, CURLMOPT_TIMERFUNCTION, upload_timer_callback);
    curl_multi_setopt(curlmupload, CURLMOPT_TIMERDATA, this);
    curluploadtimeoutreset = 0;
#endif

    curlsh = curl_share_init();
    curl_share_setopt(curlsh, CURLSHOPT_SHARE, CURL_LOCK_DATA_DNS);
    curl_share_setopt(curlsh, CURLSHOPT_SHARE, CURL_LOCK_DATA_SSL_SESSION);

    contenttypejson = curl_slist_append(NULL, "Content-Type: application/json");
    contenttypejson = curl_slist_append(contenttypejson, "Expect:");

    contenttypebinary = curl_slist_append(NULL, "Content-Type: application/octet-stream");
    contenttypebinary = curl_slist_append(contenttypebinary, "Expect:");

    proxyinflight = 0;
    ipv6requestsenabled = false;
    ipv6proxyenabled = ipv6requestsenabled;
    ipv6deactivationtime = Waiter::ds;
    waiter = NULL;
    proxyport = 0;
}

bool CurlHttpIO::ipv6available()
{
    static int ipv6_works = -1;

    if (ipv6_works != -1)
    {
        return ipv6_works;
    }

    int s = socket(PF_INET6, SOCK_DGRAM, 0);

    if (s == -1)
    {
        ipv6_works = 0;
    }
    else
    {
        ipv6_works = curlipv6;
#ifdef _WIN32
		closesocket(s);
#else
        close(s);
#endif
    }

    return ipv6_works;
}

void CurlHttpIO::filterDNSservers()
{
    string newservers;
    string serverlist;
    set<string> serverset;
    vector<string> filteredservers;
    ares_addr_node *servers;
    ares_addr_node *server;
    if (ares_get_servers(ares, &servers) == ARES_SUCCESS)
    {
        bool first = true;
        bool filtered = false;
        server = servers;
        while (server)
        {
            char straddr[INET6_ADDRSTRLEN];
            straddr[0] = 0;

            if (server->family == AF_INET6)
            {
                mega_inet_ntop(PF_INET6, &server->addr, straddr, sizeof(straddr));
            }
            else if (server->family == AF_INET)
            {
                mega_inet_ntop(PF_INET, &server->addr, straddr, sizeof(straddr));
            }
            else
            {
                LOG_warn << "Unknown IP address family: " << server->family;
            }

            if (straddr[0])
            {
                serverlist.append(straddr);
                serverlist.append(",");
            }

            if (straddr[0]
                    && serverset.find(straddr) == serverset.end()
                    && strncasecmp(straddr, "fec0:", 5)
                    && strncasecmp(straddr, "169.254.", 8))
            {
                if (!first)
                {
                    newservers.append(",");
                }

                newservers.append(straddr);
                serverset.insert(straddr);
                first = false;
            }
            else
            {
                filtered = true;
                if (!straddr[0])
                {
                    LOG_debug << "Filtering unkwnown address of DNS server";
                }
                else if (serverset.find(straddr) == serverset.end())
                {
                    serverset.insert(straddr);
                    filteredservers.push_back(straddr);
                }
            }

            server = server->next;
        }

        if (serverlist.size())
        {
            serverlist.resize(serverlist.size() - 1);
        }
        LOG_debug << "DNS servers: " << serverlist;

        if (filtered && (newservers.size() || filteredservers.size()))
        {
            for (unsigned int i = 0; i < filteredservers.size(); i++)
            {
                if (newservers.size())
                {
                    newservers.append(",");
                }

                newservers.append(filteredservers[i]);
            }

            LOG_debug << "Setting filtered DNS servers: " << newservers;
            ares_set_servers_csv(ares, newservers.c_str());
        }

        ares_free_data(servers);
    }
}

#if defined(_WIN32) && !defined(WINDOWS_PHONE)
void CurlHttpIO::addaresevents(WinWaiter *waiter)
{
    long events;

    for (unsigned int i = 0; i < aressockets.size(); i++)
    {
        if (aressockets[i].handle != WSA_INVALID_EVENT)
        {
            WSACloseEvent(aressockets[i].handle);
        }
    }

    aressockets.clear();
    ares_socket_t socks[ARES_GETSOCK_MAXNUM];
    int bitmask = ares_getsock(ares, socks, ARES_GETSOCK_MAXNUM);
    for (int i = 0; i < ARES_GETSOCK_MAXNUM; i++)
    {
        SockInfo info;

        events = 0;
        if(ARES_GETSOCK_READABLE(bitmask, i))
        {
            info.fd = socks[i];
            info.mode |= SockInfo::READ;
            events |= FD_READ;
        }

        if(ARES_GETSOCK_WRITABLE(bitmask, i))
        {
            info.fd = socks[i];
            info.mode |= SockInfo::WRITE;
            events |= FD_WRITE;
        }

        if (!info.mode)
        {
            break;
        }

        info.handle = WSACreateEvent();
        if (info.handle == WSA_INVALID_EVENT)
        {
            LOG_err << "Unable to create WSA event for cares";
        }
        else if (WSAEventSelect(info.fd, info.handle, events))
        {
            LOG_err << "Error associating cares handle " << info.fd << ": " << GetLastError();
            info.handle = WSA_INVALID_EVENT;
        }

        if (info.handle != WSA_INVALID_EVENT)
        {
            ((WinWaiter *)waiter)->addhandle(info.handle, Waiter::NEEDEXEC);
        }

        aressockets.push_back(info);
    }
}

void CurlHttpIO::addcurlevents(WinWaiter *waiter, direction_t d)
{
    std::map<int, SockInfo> &socketmap = (d == API) ? curlapisockets : ((d == GET) ? curldownloadsockets : curluploadsockets);

    long events;
    for (std::map<int, SockInfo>::iterator it = socketmap.begin(); it != socketmap.end(); it++)
    {
        SockInfo &info = it->second;

        if (!info.mode)
        {
            continue;
        }

        if (info.handle == WSA_INVALID_EVENT)
        {
            info.handle = WSACreateEvent();
            if (info.handle == WSA_INVALID_EVENT)
            {
                LOG_err << "Unable to create WSA event for curl";
                continue;
            }
        }

        events = 0;
        if (info.mode & SockInfo::READ)
        {
            events |= FD_READ;
        }

        if (info.mode & SockInfo::WRITE)
        {
            events |= FD_WRITE;
        }

        if (WSAEventSelect(info.fd, info.handle, events))
        {
            LOG_err << "Error associating curl handle " << info.fd << ": " << GetLastError();
            WSACloseEvent(info.handle);
            info.handle = WSA_INVALID_EVENT;
            continue;
        }

        ((WinWaiter *)waiter)->addhandle(info.handle, Waiter::NEEDEXEC);
    }
}

void CurlHttpIO::closecurlevents(direction_t d)
{
    std::map<int, SockInfo> &socketmap = (d == API) ? curlapisockets : ((d == GET) ? curldownloadsockets : curluploadsockets);

    for (std::map<int, SockInfo>::iterator it = socketmap.begin(); it != socketmap.end(); it++)
    {
        SockInfo &info = it->second;
        if (info.handle != WSA_INVALID_EVENT)
        {
            WSACloseEvent(info.handle);
        }
    }
    socketmap.clear();
}

void CurlHttpIO::processcurlevents(direction_t d)
{
    int dummy = 0;
    std::map<int, SockInfo> &socketmap = (d == API) ? curlapisockets : ((d == GET) ? curldownloadsockets : curluploadsockets);

    bool active = false;
    for (std::map<int, SockInfo>::iterator it = socketmap.begin(); it != socketmap.end();)
    {
        SockInfo &info = (it++)->second;
        if (!info.mode || info.handle == WSA_INVALID_EVENT)
        {
            continue;
        }

        if (WSAWaitForMultipleEvents(1, &info.handle, TRUE, 0, FALSE) == WSA_WAIT_EVENT_0)
        {
            active = true;
            WSAResetEvent(info.handle);
            curl_multi_socket_action((d == API) ? curlmapi : ((d == GET) ? curlmdownload : curlmupload),
                                     info.fd,
                                     ((info.mode & SockInfo::READ) ? CURL_CSELECT_IN : 0)
                                   | ((info.mode & SockInfo::WRITE) ? CURL_CSELECT_OUT : 0),
                                     &dummy);
            break;
        }
    }

    if (!active)
    {
        curl_multi_socket_action((d == API) ? curlmapi : ((d == GET) ? curlmdownload : curlmupload), CURL_SOCKET_TIMEOUT, 0, &dummy);
    }
}
#endif

CurlHttpIO::~CurlHttpIO()
{
    ares_destroy(ares);
    curl_multi_cleanup(curlmapi);
    curl_multi_cleanup(curlmdownload);
    curl_multi_cleanup(curlmupload);

#if defined(_WIN32) && !defined(WINDOWS_PHONE)
    for (unsigned int i = 0; i < aressockets.size(); i++)
    {
        if (aressockets[i].handle != WSA_INVALID_EVENT)
        {
            WSACloseEvent(aressockets[i].handle);
        }
    }
    aressockets.clear();
    closecurlevents(API);
    closecurlevents(GET);
    closecurlevents(PUT);
#endif

    curlMutex.lock();
    ares_library_cleanup();
    curl_global_cleanup();
    curlMutex.unlock();
}

void CurlHttpIO::setuseragent(string* u)
{
    useragent = *u;
}

void CurlHttpIO::setdnsservers(const char* servers)
{
    if (servers)
    {
        lastdnspurge = Waiter::ds + DNS_CACHE_TIMEOUT_DS / 2;
        dnscache.clear();

        dnsservers = servers;

        LOG_debug << "Using custom DNS servers: " << dnsservers;
        ares_set_servers_csv(ares, servers);
    }
}

void CurlHttpIO::disconnect()
{
    LOG_debug << "Reinitializing the network layer";

    ares_destroy(ares);
    curl_multi_cleanup(curlmapi);
    curl_multi_cleanup(curlmdownload);
    curl_multi_cleanup(curlmupload);

#if defined(_WIN32) && !defined(WINDOWS_PHONE)
    for (unsigned int i = 0; i < aressockets.size(); i++)
    {
        if (aressockets[i].handle != WSA_INVALID_EVENT)
        {
            WSACloseEvent(aressockets[i].handle);
        }
    }
    aressockets.clear();
    closecurlevents(API);
    closecurlevents(GET);
    closecurlevents(PUT);
#endif

    lastdnspurge = Waiter::ds + DNS_CACHE_TIMEOUT_DS / 2;
    dnscache.clear();

    curlmapi = curl_multi_init();
    curlmdownload = curl_multi_init();
    curlmupload = curl_multi_init();
    struct ares_options options;
    options.tries = 2;
    ares_init_options(&ares, &options, ARES_OPT_TRIES);

#if defined(_WIN32) && !defined(WINDOWS_PHONE)
    curl_multi_setopt(curlmapi, CURLMOPT_SOCKETFUNCTION, api_socket_callback);
    curl_multi_setopt(curlmapi, CURLMOPT_SOCKETDATA, this);
    curl_multi_setopt(curlmapi, CURLMOPT_TIMERFUNCTION, api_timer_callback);
    curl_multi_setopt(curlmapi, CURLMOPT_TIMERDATA, this);
    curlapitimeoutreset = 0;

    curl_multi_setopt(curlmdownload, CURLMOPT_SOCKETFUNCTION, download_socket_callback);
    curl_multi_setopt(curlmdownload, CURLMOPT_SOCKETDATA, this);
    curl_multi_setopt(curlmdownload, CURLMOPT_TIMERFUNCTION, download_timer_callback);
    curl_multi_setopt(curlmdownload, CURLMOPT_TIMERDATA, this);
    curldownloadtimeoutreset = 0;

    curl_multi_setopt(curlmupload, CURLMOPT_SOCKETFUNCTION, upload_socket_callback);
    curl_multi_setopt(curlmupload, CURLMOPT_SOCKETDATA, this);
    curl_multi_setopt(curlmupload, CURLMOPT_TIMERFUNCTION, upload_timer_callback);
    curl_multi_setopt(curlmupload, CURLMOPT_TIMERDATA, this);
    curluploadtimeoutreset = 0;
#endif

    if (dnsservers.size())
    {
        LOG_debug << "Using custom DNS servers: " << dnsservers;
        ares_set_servers_csv(ares, dnsservers.c_str());
    }
    else
    {
        filterDNSservers();
    }

    if (proxyurl.size() && !proxyip.size())
    {
        LOG_debug << "Unresolved proxy name. Resolving...";
        request_proxy_ip();
    }
}

bool CurlHttpIO::setmaxdownloadspeed(m_off_t bpslimit)
{
    maxdownloadspeed = bpslimit;
    return true;
}

bool CurlHttpIO::setmaxuploadspeed(m_off_t bpslimit)
{
    maxuploadspeed = bpslimit;
    return true;
}

m_off_t CurlHttpIO::getmaxdownloadspeed()
{
    return maxdownloadspeed;
}

m_off_t CurlHttpIO::getmaxuploadspeed()
{
    return maxuploadspeed;
}

// wake up from cURL I/O
void CurlHttpIO::addevents(Waiter* w, int)
{
    waiter = (WAIT_CLASS*)w;
    long curltimeoutms = -1;

#if !defined(_WIN32) || defined(WINDOWS_PHONE)
    int t;
    long ms;
    t = ares_fds(ares, &waiter->rfds, &waiter->wfds);
    waiter->bumpmaxfd(t);
#else
    addaresevents(waiter);
#endif

#if !defined(_WIN32) || defined(WINDOWS_PHONE)
        curl_multi_fdset(curlmapi, &waiter->rfds, &waiter->wfds, &waiter->efds, &t);
        waiter->bumpmaxfd(t);

        ms = -1;
        curl_multi_timeout(curlmapi, &ms);
        if (curltimeoutms < 0 || (ms >= 0 && curltimeoutms > ms))
        {
            curltimeoutms = ms;
        }
#else
        addcurlevents(waiter, API);
        if (curlapitimeoutreset)
        {
            m_time_t ds = curlapitimeoutreset - Waiter::ds;
            if (ds <= 0)
            {
                curltimeoutms = 0;
                curlapitimeoutreset = 0;
                LOG_debug << "Disabling cURL timeout for API requests";
            }
            else
            {
                if (curltimeoutms < 0 || curltimeoutms > ds * 100)
                {
                    curltimeoutms = ds * 100;
                }
            }
        }
#endif

    if (!maxdownloadspeed || maxdownloadspeed > downloadSpeed)
    {
#if !defined(_WIN32) || defined(WINDOWS_PHONE)
        curl_multi_fdset(curlmdownload, &waiter->rfds, &waiter->wfds, &waiter->efds, &t);
        waiter->bumpmaxfd(t);

        ms = -1;
        curl_multi_timeout(curlmdownload, &ms);
        if (curltimeoutms < 0 || (ms >= 0 && curltimeoutms > ms))
        {
            curltimeoutms = ms;
        }
#else
        addcurlevents(waiter, GET);
        if (curldownloadtimeoutreset)
        {
            m_time_t ds = curldownloadtimeoutreset - Waiter::ds;
            if (ds <= 0)
            {
                curltimeoutms = 0;
                curldownloadtimeoutreset = 0;
                LOG_debug << "Disabling cURL timeout for downloads";
            }
            else
            {
                if (curltimeoutms < 0 || curltimeoutms > ds * 100)
                {
                    curltimeoutms = ds * 100;
                }
            }
        }
#endif
    }
    else
    {
        m_off_t excess = downloadSpeed - maxdownloadspeed;
        long ms = 1000 * excess / maxdownloadspeed;
        if (curltimeoutms < 0 || ms < curltimeoutms)
        {
            curltimeoutms = ms;
        }

        if (curltimeoutms > MAX_SPEED_CONTROL_TIMEOUT_MS)
        {
            curltimeoutms = MAX_SPEED_CONTROL_TIMEOUT_MS;
        }
    }

    if (!maxuploadspeed || maxuploadspeed > uploadSpeed)
    {
#if !defined(_WIN32) || defined(WINDOWS_PHONE)
        curl_multi_fdset(curlmupload, &waiter->rfds, &waiter->wfds, &waiter->efds, &t);
        waiter->bumpmaxfd(t);

        ms = -1;
        curl_multi_timeout(curlmupload, &ms);
        if (curltimeoutms < 0 || (ms >= 0 && curltimeoutms > ms))
        {
            curltimeoutms = ms;
        }
#else
        addcurlevents(waiter, PUT);
        if (curluploadtimeoutreset)
        {
            m_time_t ds = curluploadtimeoutreset - Waiter::ds;
            if (ds <= 0)
            {
                curltimeoutms = 0;
                curluploadtimeoutreset = 0;
                LOG_debug << "Disabling cURL timeout for uploads";
            }
            else
            {
                if (curltimeoutms < 0 || curltimeoutms > ds * 100)
                {
                    curltimeoutms = ds * 100;
                }
            }
        }
#endif
    }
    else
    {
        m_off_t excess = uploadSpeed - maxuploadspeed;
        long ms = 1000 * excess / maxuploadspeed;
        if (curltimeoutms < 0 || ms < curltimeoutms)
        {
            curltimeoutms = ms;
        }

        if (curltimeoutms > MAX_SPEED_CONTROL_TIMEOUT_MS)
        {
            curltimeoutms = MAX_SPEED_CONTROL_TIMEOUT_MS;
        }
    }

    if (curltimeoutms >= 0)
    {<commit_msg>Wake up the SDK to calculate the transfer speed<commit_after>/**
 * @file posix/net.cpp
 * @brief POSIX network access layer (using cURL + c-ares)
 *
 * (c) 2013-2014 by Mega Limited, Auckland, New Zealand
 *
 * This file is part of the MEGA SDK - Client Access Engine.
 *
 * Applications using the MEGA API must present a valid application key
 * and comply with the the rules set forth in the Terms of Service.
 *
 * The MEGA SDK is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
 *
 * @copyright Simplified (2-clause) BSD License.
 *
 * You should have received a copy of the license along with this
 * program.
 */

#include "mega.h"
#include "mega/posix/meganet.h"
#include "mega/logging.h"

#define IPV6_RETRY_INTERVAL_DS 72000
#define DNS_CACHE_TIMEOUT_DS 18000
#define MAX_SPEED_CONTROL_TIMEOUT_MS 500

namespace mega {

MUTEX_CLASS CurlHttpIO::curlMutex(false);

#if !defined(USE_CURL_PUBLIC_KEY_PINNING) || defined(WINDOWS_PHONE)

MUTEX_CLASS **CurlHttpIO::sslMutexes = NULL;
void CurlHttpIO::locking_function(int mode, int lockNumber, const char *, int)
{
    MUTEX_CLASS *mutex = sslMutexes[lockNumber];
    if (mutex == NULL)
    {
        mutex = new MUTEX_CLASS(true);
        sslMutexes[lockNumber] = mutex;
    }

    if (mode & CRYPTO_LOCK)
    {
        mutex->lock();
    }
    else
    {
        mutex->unlock();
    }
}

#if OPENSSL_VERSION_NUMBER >= 0x10000000
void CurlHttpIO::id_function(CRYPTO_THREADID* id)
{
    CRYPTO_THREADID_set_pointer(id, (void *)THREAD_CLASS::currentThreadId());
}
#else
unsigned long CurlHttpIO::id_function()
{
    return THREAD_CLASS::currentThreadId();
}
#endif

#endif

CurlHttpIO::CurlHttpIO()
{
    curl_version_info_data* data = curl_version_info(CURLVERSION_NOW);
    string curlssl = data->ssl_version;
    std::transform(curlssl.begin(), curlssl.end(), curlssl.begin(), ::tolower);

#if !defined(USE_CURL_PUBLIC_KEY_PINNING) || defined(WINDOWS_PHONE)
    if (!strstr(curlssl.c_str(), "openssl"))
    {
        LOG_fatal << "cURL built without OpenSSL support. Aborting.";
        exit(EXIT_FAILURE);
    }
#endif

    int i;

    for (i = 0; data->protocols[i]; i++)
    {
        if (strstr(data->protocols[i], "http"))
        {
            break;
        }
    }

    if (!data->protocols[i] || !(data->features & CURL_VERSION_SSL))
    {
        LOG_fatal << "cURL built without HTTP/HTTPS support. Aborting.";
        exit(EXIT_FAILURE);
    }

    curlipv6 = data->features & CURL_VERSION_IPV6;
    LOG_debug << "IPv6 enabled: " << curlipv6;

    dnsok = false;
    reset = false;
    statechange = false;
    maxdownloadspeed = 0;
    maxuploadspeed = 0;

    WAIT_CLASS::bumpds();
    lastdnspurge = Waiter::ds + DNS_CACHE_TIMEOUT_DS / 2;

    curlMutex.lock();

#if !defined(USE_CURL_PUBLIC_KEY_PINNING) || defined(WINDOWS_PHONE)

    if (!CRYPTO_get_locking_callback()
#if OPENSSL_VERSION_NUMBER >= 0x10000000
        && !CRYPTO_THREADID_get_callback())
#else
        && !CRYPTO_get_id_callback())
#endif
    {
        LOG_debug << "Initializing OpenSSL locking callbacks";
        int numLocks = CRYPTO_num_locks();
        sslMutexes = new MUTEX_CLASS*[numLocks];
        memset(sslMutexes, 0, numLocks * sizeof(MUTEX_CLASS*));
#if OPENSSL_VERSION_NUMBER >= 0x10000000
        CRYPTO_THREADID_set_callback(CurlHttpIO::id_function);
#else
        CRYPTO_set_id_callback(CurlHttpIO::id_function);
#endif
        CRYPTO_set_locking_callback(CurlHttpIO::locking_function);
    }

#endif

    curl_global_init(CURL_GLOBAL_DEFAULT);
    ares_library_init(ARES_LIB_INIT_ALL);
    curlMutex.unlock();

    curlmapi = curl_multi_init();
    curlmdownload = curl_multi_init();
    curlmupload = curl_multi_init();

    struct ares_options options;
    options.tries = 2;
    ares_init_options(&ares, &options, ARES_OPT_TRIES);
    filterDNSservers();

#if defined(_WIN32) && !defined(WINDOWS_PHONE)
    curl_multi_setopt(curlmapi, CURLMOPT_SOCKETFUNCTION, api_socket_callback);
    curl_multi_setopt(curlmapi, CURLMOPT_SOCKETDATA, this);
    curl_multi_setopt(curlmapi, CURLMOPT_TIMERFUNCTION, api_timer_callback);
    curl_multi_setopt(curlmapi, CURLMOPT_TIMERDATA, this);
    curlapitimeoutreset = 0;

    curl_multi_setopt(curlmdownload, CURLMOPT_SOCKETFUNCTION, download_socket_callback);
    curl_multi_setopt(curlmdownload, CURLMOPT_SOCKETDATA, this);
    curl_multi_setopt(curlmdownload, CURLMOPT_TIMERFUNCTION, download_timer_callback);
    curl_multi_setopt(curlmdownload, CURLMOPT_TIMERDATA, this);
    curldownloadtimeoutreset = 0;

    curl_multi_setopt(curlmupload, CURLMOPT_SOCKETFUNCTION, upload_socket_callback);
    curl_multi_setopt(curlmupload, CURLMOPT_SOCKETDATA, this);
    curl_multi_setopt(curlmupload, CURLMOPT_TIMERFUNCTION, upload_timer_callback);
    curl_multi_setopt(curlmupload, CURLMOPT_TIMERDATA, this);
    curluploadtimeoutreset = 0;
#endif

    curlsh = curl_share_init();
    curl_share_setopt(curlsh, CURLSHOPT_SHARE, CURL_LOCK_DATA_DNS);
    curl_share_setopt(curlsh, CURLSHOPT_SHARE, CURL_LOCK_DATA_SSL_SESSION);

    contenttypejson = curl_slist_append(NULL, "Content-Type: application/json");
    contenttypejson = curl_slist_append(contenttypejson, "Expect:");

    contenttypebinary = curl_slist_append(NULL, "Content-Type: application/octet-stream");
    contenttypebinary = curl_slist_append(contenttypebinary, "Expect:");

    proxyinflight = 0;
    ipv6requestsenabled = false;
    ipv6proxyenabled = ipv6requestsenabled;
    ipv6deactivationtime = Waiter::ds;
    waiter = NULL;
    proxyport = 0;
}

bool CurlHttpIO::ipv6available()
{
    static int ipv6_works = -1;

    if (ipv6_works != -1)
    {
        return ipv6_works;
    }

    int s = socket(PF_INET6, SOCK_DGRAM, 0);

    if (s == -1)
    {
        ipv6_works = 0;
    }
    else
    {
        ipv6_works = curlipv6;
#ifdef _WIN32
		closesocket(s);
#else
        close(s);
#endif
    }

    return ipv6_works;
}

void CurlHttpIO::filterDNSservers()
{
    string newservers;
    string serverlist;
    set<string> serverset;
    vector<string> filteredservers;
    ares_addr_node *servers;
    ares_addr_node *server;
    if (ares_get_servers(ares, &servers) == ARES_SUCCESS)
    {
        bool first = true;
        bool filtered = false;
        server = servers;
        while (server)
        {
            char straddr[INET6_ADDRSTRLEN];
            straddr[0] = 0;

            if (server->family == AF_INET6)
            {
                mega_inet_ntop(PF_INET6, &server->addr, straddr, sizeof(straddr));
            }
            else if (server->family == AF_INET)
            {
                mega_inet_ntop(PF_INET, &server->addr, straddr, sizeof(straddr));
            }
            else
            {
                LOG_warn << "Unknown IP address family: " << server->family;
            }

            if (straddr[0])
            {
                serverlist.append(straddr);
                serverlist.append(",");
            }

            if (straddr[0]
                    && serverset.find(straddr) == serverset.end()
                    && strncasecmp(straddr, "fec0:", 5)
                    && strncasecmp(straddr, "169.254.", 8))
            {
                if (!first)
                {
                    newservers.append(",");
                }

                newservers.append(straddr);
                serverset.insert(straddr);
                first = false;
            }
            else
            {
                filtered = true;
                if (!straddr[0])
                {
                    LOG_debug << "Filtering unkwnown address of DNS server";
                }
                else if (serverset.find(straddr) == serverset.end())
                {
                    serverset.insert(straddr);
                    filteredservers.push_back(straddr);
                }
            }

            server = server->next;
        }

        if (serverlist.size())
        {
            serverlist.resize(serverlist.size() - 1);
        }
        LOG_debug << "DNS servers: " << serverlist;

        if (filtered && (newservers.size() || filteredservers.size()))
        {
            for (unsigned int i = 0; i < filteredservers.size(); i++)
            {
                if (newservers.size())
                {
                    newservers.append(",");
                }

                newservers.append(filteredservers[i]);
            }

            LOG_debug << "Setting filtered DNS servers: " << newservers;
            ares_set_servers_csv(ares, newservers.c_str());
        }

        ares_free_data(servers);
    }
}

#if defined(_WIN32) && !defined(WINDOWS_PHONE)
void CurlHttpIO::addaresevents(WinWaiter *waiter)
{
    long events;

    for (unsigned int i = 0; i < aressockets.size(); i++)
    {
        if (aressockets[i].handle != WSA_INVALID_EVENT)
        {
            WSACloseEvent(aressockets[i].handle);
        }
    }

    aressockets.clear();
    ares_socket_t socks[ARES_GETSOCK_MAXNUM];
    int bitmask = ares_getsock(ares, socks, ARES_GETSOCK_MAXNUM);
    for (int i = 0; i < ARES_GETSOCK_MAXNUM; i++)
    {
        SockInfo info;

        events = 0;
        if(ARES_GETSOCK_READABLE(bitmask, i))
        {
            info.fd = socks[i];
            info.mode |= SockInfo::READ;
            events |= FD_READ;
        }

        if(ARES_GETSOCK_WRITABLE(bitmask, i))
        {
            info.fd = socks[i];
            info.mode |= SockInfo::WRITE;
            events |= FD_WRITE;
        }

        if (!info.mode)
        {
            break;
        }

        info.handle = WSACreateEvent();
        if (info.handle == WSA_INVALID_EVENT)
        {
            LOG_err << "Unable to create WSA event for cares";
        }
        else if (WSAEventSelect(info.fd, info.handle, events))
        {
            LOG_err << "Error associating cares handle " << info.fd << ": " << GetLastError();
            info.handle = WSA_INVALID_EVENT;
        }

        if (info.handle != WSA_INVALID_EVENT)
        {
            ((WinWaiter *)waiter)->addhandle(info.handle, Waiter::NEEDEXEC);
        }

        aressockets.push_back(info);
    }
}

void CurlHttpIO::addcurlevents(WinWaiter *waiter, direction_t d)
{
    std::map<int, SockInfo> &socketmap = (d == API) ? curlapisockets : ((d == GET) ? curldownloadsockets : curluploadsockets);

    long events;
    for (std::map<int, SockInfo>::iterator it = socketmap.begin(); it != socketmap.end(); it++)
    {
        SockInfo &info = it->second;

        if (!info.mode)
        {
            continue;
        }

        if (info.handle == WSA_INVALID_EVENT)
        {
            info.handle = WSACreateEvent();
            if (info.handle == WSA_INVALID_EVENT)
            {
                LOG_err << "Unable to create WSA event for curl";
                continue;
            }
        }

        events = 0;
        if (info.mode & SockInfo::READ)
        {
            events |= FD_READ;
        }

        if (info.mode & SockInfo::WRITE)
        {
            events |= FD_WRITE;
        }

        if (WSAEventSelect(info.fd, info.handle, events))
        {
            LOG_err << "Error associating curl handle " << info.fd << ": " << GetLastError();
            WSACloseEvent(info.handle);
            info.handle = WSA_INVALID_EVENT;
            continue;
        }

        ((WinWaiter *)waiter)->addhandle(info.handle, Waiter::NEEDEXEC);
    }
}

void CurlHttpIO::closecurlevents(direction_t d)
{
    std::map<int, SockInfo> &socketmap = (d == API) ? curlapisockets : ((d == GET) ? curldownloadsockets : curluploadsockets);

    for (std::map<int, SockInfo>::iterator it = socketmap.begin(); it != socketmap.end(); it++)
    {
        SockInfo &info = it->second;
        if (info.handle != WSA_INVALID_EVENT)
        {
            WSACloseEvent(info.handle);
        }
    }
    socketmap.clear();
}

void CurlHttpIO::processcurlevents(direction_t d)
{
    int dummy = 0;
    std::map<int, SockInfo> &socketmap = (d == API) ? curlapisockets : ((d == GET) ? curldownloadsockets : curluploadsockets);

    bool active = false;
    for (std::map<int, SockInfo>::iterator it = socketmap.begin(); it != socketmap.end();)
    {
        SockInfo &info = (it++)->second;
        if (!info.mode || info.handle == WSA_INVALID_EVENT)
        {
            continue;
        }

        if (WSAWaitForMultipleEvents(1, &info.handle, TRUE, 0, FALSE) == WSA_WAIT_EVENT_0)
        {
            active = true;
            WSAResetEvent(info.handle);
            curl_multi_socket_action((d == API) ? curlmapi : ((d == GET) ? curlmdownload : curlmupload),
                                     info.fd,
                                     ((info.mode & SockInfo::READ) ? CURL_CSELECT_IN : 0)
                                   | ((info.mode & SockInfo::WRITE) ? CURL_CSELECT_OUT : 0),
                                     &dummy);
            break;
        }
    }

    if (!active)
    {
        curl_multi_socket_action((d == API) ? curlmapi : ((d == GET) ? curlmdownload : curlmupload), CURL_SOCKET_TIMEOUT, 0, &dummy);
    }
}
#endif

CurlHttpIO::~CurlHttpIO()
{
    ares_destroy(ares);
    curl_multi_cleanup(curlmapi);
    curl_multi_cleanup(curlmdownload);
    curl_multi_cleanup(curlmupload);

#if defined(_WIN32) && !defined(WINDOWS_PHONE)
    for (unsigned int i = 0; i < aressockets.size(); i++)
    {
        if (aressockets[i].handle != WSA_INVALID_EVENT)
        {
            WSACloseEvent(aressockets[i].handle);
        }
    }
    aressockets.clear();
    closecurlevents(API);
    closecurlevents(GET);
    closecurlevents(PUT);
#endif

    curlMutex.lock();
    ares_library_cleanup();
    curl_global_cleanup();
    curlMutex.unlock();
}

void CurlHttpIO::setuseragent(string* u)
{
    useragent = *u;
}

void CurlHttpIO::setdnsservers(const char* servers)
{
    if (servers)
    {
        lastdnspurge = Waiter::ds + DNS_CACHE_TIMEOUT_DS / 2;
        dnscache.clear();

        dnsservers = servers;

        LOG_debug << "Using custom DNS servers: " << dnsservers;
        ares_set_servers_csv(ares, servers);
    }
}

void CurlHttpIO::disconnect()
{
    LOG_debug << "Reinitializing the network layer";

    ares_destroy(ares);
    curl_multi_cleanup(curlmapi);
    curl_multi_cleanup(curlmdownload);
    curl_multi_cleanup(curlmupload);

#if defined(_WIN32) && !defined(WINDOWS_PHONE)
    for (unsigned int i = 0; i < aressockets.size(); i++)
    {
        if (aressockets[i].handle != WSA_INVALID_EVENT)
        {
            WSACloseEvent(aressockets[i].handle);
        }
    }
    aressockets.clear();
    closecurlevents(API);
    closecurlevents(GET);
    closecurlevents(PUT);
#endif

    lastdnspurge = Waiter::ds + DNS_CACHE_TIMEOUT_DS / 2;
    dnscache.clear();

    curlmapi = curl_multi_init();
    curlmdownload = curl_multi_init();
    curlmupload = curl_multi_init();
    struct ares_options options;
    options.tries = 2;
    ares_init_options(&ares, &options, ARES_OPT_TRIES);

#if defined(_WIN32) && !defined(WINDOWS_PHONE)
    curl_multi_setopt(curlmapi, CURLMOPT_SOCKETFUNCTION, api_socket_callback);
    curl_multi_setopt(curlmapi, CURLMOPT_SOCKETDATA, this);
    curl_multi_setopt(curlmapi, CURLMOPT_TIMERFUNCTION, api_timer_callback);
    curl_multi_setopt(curlmapi, CURLMOPT_TIMERDATA, this);
    curlapitimeoutreset = 0;

    curl_multi_setopt(curlmdownload, CURLMOPT_SOCKETFUNCTION, download_socket_callback);
    curl_multi_setopt(curlmdownload, CURLMOPT_SOCKETDATA, this);
    curl_multi_setopt(curlmdownload, CURLMOPT_TIMERFUNCTION, download_timer_callback);
    curl_multi_setopt(curlmdownload, CURLMOPT_TIMERDATA, this);
    curldownloadtimeoutreset = 0;

    curl_multi_setopt(curlmupload, CURLMOPT_SOCKETFUNCTION, upload_socket_callback);
    curl_multi_setopt(curlmupload, CURLMOPT_SOCKETDATA, this);
    curl_multi_setopt(curlmupload, CURLMOPT_TIMERFUNCTION, upload_timer_callback);
    curl_multi_setopt(curlmupload, CURLMOPT_TIMERDATA, this);
    curluploadtimeoutreset = 0;
#endif

    if (dnsservers.size())
    {
        LOG_debug << "Using custom DNS servers: " << dnsservers;
        ares_set_servers_csv(ares, dnsservers.c_str());
    }
    else
    {
        filterDNSservers();
    }

    if (proxyurl.size() && !proxyip.size())
    {
        LOG_debug << "Unresolved proxy name. Resolving...";
        request_proxy_ip();
    }
}

bool CurlHttpIO::setmaxdownloadspeed(m_off_t bpslimit)
{
    maxdownloadspeed = bpslimit;
    return true;
}

bool CurlHttpIO::setmaxuploadspeed(m_off_t bpslimit)
{
    maxuploadspeed = bpslimit;
    return true;
}

m_off_t CurlHttpIO::getmaxdownloadspeed()
{
    return maxdownloadspeed;
}

m_off_t CurlHttpIO::getmaxuploadspeed()
{
    return maxuploadspeed;
}

// wake up from cURL I/O
void CurlHttpIO::addevents(Waiter* w, int)
{
    waiter = (WAIT_CLASS*)w;
    long curltimeoutms = -1;

#if !defined(_WIN32) || defined(WINDOWS_PHONE)
    int t;
    long ms;
    t = ares_fds(ares, &waiter->rfds, &waiter->wfds);
    waiter->bumpmaxfd(t);
#else
    addaresevents(waiter);
#endif

#if !defined(_WIN32) || defined(WINDOWS_PHONE)
        curl_multi_fdset(curlmapi, &waiter->rfds, &waiter->wfds, &waiter->efds, &t);
        waiter->bumpmaxfd(t);

        ms = -1;
        curl_multi_timeout(curlmapi, &ms);
        if (curltimeoutms < 0 || (ms >= 0 && curltimeoutms > ms))
        {
            curltimeoutms = ms;
        }
#else
        addcurlevents(waiter, API);
        if (curlapitimeoutreset)
        {
            m_time_t ds = curlapitimeoutreset - Waiter::ds;
            if (ds <= 0)
            {
                curltimeoutms = 0;
                curlapitimeoutreset = 0;
                LOG_debug << "Disabling cURL timeout for API requests";
            }
            else
            {
                if (curltimeoutms < 0 || curltimeoutms > ds * 100)
                {
                    curltimeoutms = ds * 100;
                }
            }
        }
#endif

    if (!maxdownloadspeed || maxdownloadspeed > downloadSpeed)
    {
#if !defined(_WIN32) || defined(WINDOWS_PHONE)
        curl_multi_fdset(curlmdownload, &waiter->rfds, &waiter->wfds, &waiter->efds, &t);
        waiter->bumpmaxfd(t);

        ms = -1;
        curl_multi_timeout(curlmdownload, &ms);
        if (curltimeoutms < 0 || (ms >= 0 && curltimeoutms > ms))
        {
            curltimeoutms = ms;
        }
#else
        addcurlevents(waiter, GET);
        if (curldownloadtimeoutreset)
        {
            m_time_t ds = curldownloadtimeoutreset - Waiter::ds;
            if (ds <= 0)
            {
                curltimeoutms = 0;
                curldownloadtimeoutreset = 0;
                LOG_debug << "Disabling cURL timeout for downloads";
            }
            else
            {
                if (curltimeoutms < 0 || curltimeoutms > ds * 100)
                {
                    curltimeoutms = ds * 100;
                }
            }
        }
#endif
    }
    else
    {
        m_off_t excess = downloadSpeed - maxdownloadspeed;
        long ms = 1000 * excess / maxdownloadspeed;
        if (curltimeoutms < 0 || ms < curltimeoutms)
        {
            curltimeoutms = ms;
        }
    }

    if (!maxuploadspeed || maxuploadspeed > uploadSpeed)
    {
#if !defined(_WIN32) || defined(WINDOWS_PHONE)
        curl_multi_fdset(curlmupload, &waiter->rfds, &waiter->wfds, &waiter->efds, &t);
        waiter->bumpmaxfd(t);

        ms = -1;
        curl_multi_timeout(curlmupload, &ms);
        if (curltimeoutms < 0 || (ms >= 0 && curltimeoutms > ms))
        {
            curltimeoutms = ms;
        }
#else
        addcurlevents(waiter, PUT);
        if (curluploadtimeoutreset)
        {
            m_time_t ds = curluploadtimeoutreset - Waiter::ds;
            if (ds <= 0)
            {
                curltimeoutms = 0;
                curluploadtimeoutreset = 0;
                LOG_debug << "Disabling cURL timeout for uploads";
            }
            else
            {
                if (curltimeoutms < 0 || curltimeoutms > ds * 100)
                {
                    curltimeoutms = ds * 100;
                }
            }
        }
#endif
    }
    else
    {
        m_off_t excess = uploadSpeed - maxuploadspeed;
        long ms = 1000 * excess / maxuploadspeed;
        if (curltimeoutms < 0 || ms < curltimeoutms)
        {
            curltimeoutms = ms;
        }
    }

    if ((curltimeoutms < 0 || curltimeoutms > MAX_SPEED_CONTROL_TIMEOUT_MS)
            && (downloadSpeed || uploadSpeed))
    {
        curltimeoutms = MAX_SPEED_CONTROL_TIMEOUT_MS;
    }

    if (curltimeoutms >= 0)
    {
{'fraction_non_alphanumeric': 0.11258086318431978, 'fraction_numerical': 0.009293231630781114, 'mean_word_length': 2.29455268389662, 'pattern_counts': {'":': 4, '<': 180, '<?xml version=': 0, '>': 120, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 6, 'word_list_counts': {'cursed_substrings.json': 89, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7613995', 'n_tokens_mistral': 14184, 'n_tokens_neox': 13409, 'n_words': 2513}