#!/usr/bin/python

#

# CDDL HEADER START

#

# The contents of this file are subject to the terms of the

# Common Development and Distribution License (the "License").

# You may not use this file except in compliance with the License.

#

# You can obtain a copy of the license at usr/src/OPENSOLARIS.LICENSE

# or http://www.opensolaris.org/os/licensing.

# See the License for the specific language governing permissions

# and limitations under the License.

#

# When distributing Covered Code, include this CDDL HEADER in each

# file and include the License file at usr/src/OPENSOLARIS.LICENSE.

# If applicable, add the following below this CDDL HEADER, with the

# fields enclosed by brackets "[]" replaced with your own identifying

# information: Portions Copyright [yyyy] [name of copyright owner]

#

# CDDL HEADER END

#

#

# Copyright (c) 2009, 2010, Oracle and/or its affiliates. All rights reserved.

#

import errno

import httplib

import os

import pycurl

import time

import urlparse

# Need to ignore SIGPIPE if using pycurl in NOSIGNAL mode.

try:

        import signal

        if hasattr(signal, "SIGPIPE"):

                signal.signal(signal.SIGPIPE, signal.SIG_IGN)

except ImportError:

        pass

import pkg.client.api_errors            as api_errors

import pkg.client.transport.exception   as tx

import pkg.client.transport.fileobj     as fileobj

from collections        import deque

from pkg.client         import global_settings

pipelined_protocols = ()

response_protocols = ("ftp", "http", "https")

class TransportEngine(object):

        """This is an abstract class.  It shouldn't implement any

        of the methods that it contains.  Leave that to transport-specific

        implementations."""

class CurlTransportEngine(TransportEngine):

        """Concrete class of TransportEngine for libcurl transport."""

        def __init__(self, transport, max_conn=20):

                # Backpointer to transport object

                self.__xport = transport

                # Curl handles

                self.__mhandle = pycurl.CurlMulti()

                self.__chandles = []

                self.__active_handles = 0

                self.__max_handles = max_conn

                # Request queue

                self.__req_q = deque()

                # List of failures

                self.__failures = []

                # List of URLs successfully transferred

                self.__success = []

                # List of Orphaned URLs.

                self.__orphans = set()

                # Set default file buffer size at 128k, callers override

                # this setting after looking at VFS block size.

                self.__file_bufsz = 131072

                # Header bits and pieces

                self.__user_agent = None

                self.__common_header = {}

                self.__last_stall_check = 0

                # Set options on multi-handle

                self.__mhandle.setopt(pycurl.M_PIPELINING, 0)

                # initialize easy handles

                for i in range(self.__max_handles):

                        eh = pycurl.Curl()

                        eh.url = None

                        eh.repourl = None

                        eh.fobj = None

                        eh.r_fobj = None

                        eh.filepath = None

                        eh.success = False

                        eh.fileprog = None

                        eh.filetime = -1

                        eh.starttime = -1

                        eh.uuid = None

                        self.__chandles.append(eh)

                # copy handles into handle freelist

                self.__freehandles = self.__chandles[:]

        def __call_perform(self):

                """An internal method that invokes the multi-handle's

                perform method."""

                while 1:

                        ret, active_handles = self.__mhandle.perform()

                        if ret != pycurl.E_CALL_MULTI_PERFORM:

                                break

                self.__active_handles = active_handles

                return ret

        def add_url(self, url, filepath=None, writefunc=None, header=None,

            progclass=None, progtrack=None, sslcert=None, sslkey=None,

            repourl=None, compressible=False, sock_path=None, failonerror=True):

                """Add a URL to the transport engine.  Caller must supply

                either a filepath where the file should be downloaded,

                or a callback to a function that will peform the write.

                It may also optionally supply header information

                in a dictionary.  If the caller has a ProgressTracker,

                it should pass the tracker in progtrack.  The caller should

                also supply a class that wraps the tracker in progclass."""

                t = TransportRequest(url, filepath=filepath,

                    writefunc=writefunc, header=header, progclass=progclass,

                    progtrack=progtrack, sslcert=sslcert, sslkey=sslkey,

                    repourl=repourl, compressible=compressible,

                    sock_path=sock_path, failonerror=failonerror)

                self.__req_q.appendleft(t)

        def __check_for_stalls(self):

                """In some situations, libcurl can get itself

                tied in a knot, and fail to make progress.  Check that the

                active handles are making progress.  If none of the active

                handles have downloaded any content for the timeout period,

                reset the transport and generate exceptions for the failed

                requests."""

                current_time = time.time()

                timeout = global_settings.PKG_CLIENT_LOWSPEED_TIMEOUT

                time_list = []

                size_list = []

                failures = []

                q_hdls = [

                    hdl for hdl in self.__chandles

                    if hdl not in self.__freehandles

                ]

                # time.time() is based upon system clock.  Check that

                # our time hasn't been set backwards.  If time is set forward,

                # we'll have to expire the handles.  There's no way to detect

                # this until python properly implements gethrtime().  Solaris

                # implementations of time.clock() appear broken.

                for h in q_hdls:

                        time_elapsed = current_time - h.starttime

                        if time_elapsed < 0:

                                h.starttime = current_time

                                time_elapsed = 0

                        size_xfrd = h.getinfo(pycurl.SIZE_DOWNLOAD) + \

                            h.getinfo(pycurl.SIZE_UPLOAD)

                        time_list.append(time_elapsed)

                        size_list.append(size_xfrd)

                # If timeout is smaller than smallest elapsed time,

                # and no data has been transferred, abort.

                if timeout < min(time_list) and max(size_list) == 0:

                        for h in q_hdls:

                                url = h.url

                                uuid = h.uuid

                                urlstem = h.repourl

                                ex = tx.TransportStallError(url,

                                    repourl=urlstem, uuid=uuid)

                                self.__mhandle.remove_handle(h)

                                self.__teardown_handle(h)

                                self.__freehandles.append(h)

                                failures.append(ex)

                self.__failures.extend(failures)

        def __cleanup_requests(self):

                """Cleanup handles that have finished their request.

                Return the handles to the freelist.  Generate any

                relevant error information."""

                count, good, bad = self.__mhandle.info_read()

                failures = self.__failures

                success = self.__success

                done_handles = []

                ex_to_raise = None

                visited_repos = set()

                errors_seen = 0

                for h, en, em in bad:

                        # Get statistics for each handle.

                        repostats = self.__xport.stats[h.repourl]

                        visited_repos.add(repostats)

                        repostats.record_tx()

                        nbytes = h.getinfo(pycurl.SIZE_DOWNLOAD)

                        seconds = h.getinfo(pycurl.TOTAL_TIME)

                        conn_count = h.getinfo(pycurl.NUM_CONNECTS)

                        conn_time = h.getinfo(pycurl.CONNECT_TIME)

                        url = h.url

                        uuid = h.uuid

                        urlstem = h.repourl

                        proto = urlparse.urlsplit(url)[0]

                        # When using pipelined operations, libcurl tracks the

                        # amount of time taken for the entire pipelined request

                        # as opposed to just the amount of time for a single

                        # file in the pipeline.  So, if the connection time is 0

                        # for a request using http(s), then it was pipelined and

                        # the total time must be obtained by subtracting the

                        # time the transfer of the individual request started

                        # from the total time.

                        if conn_time == 0 and proto in pipelined_protocols:

                                # Only performing this subtraction when the

                                # conn_time is 0 allows the first request in

                                # the pipeline to properly include connection

                                # time, etc. to initiate the transfer.

                                seconds -= h.getinfo(pycurl.STARTTRANSFER_TIME)

                        elif conn_time > 0:

                                seconds -= conn_time

                        # Sometimes libcurl will report no transfer time.

                        # In that case, just use starttransfer time if it's

                        # non-zero.

                        if seconds < 0:

                                seconds = h.getinfo(pycurl.STARTTRANSFER_TIME)

                        repostats.record_progress(nbytes, seconds)

                        # Only count connections if the connection time is

                        # positive for http(s); for all other protocols,

                        # record the connection regardless.

                        if conn_count > 0 and conn_time > 0:

                                repostats.record_connection(conn_time)

                        respcode = h.getinfo(pycurl.RESPONSE_CODE)

                        # If we were cancelled, raise an API error.

                        # Otherwise fall through to transport's exception

                        # generation.

                        if en == pycurl.E_ABORTED_BY_CALLBACK:

                                ex = None

                                ex_to_raise = api_errors.CanceledException

                        elif en in (pycurl.E_HTTP_RETURNED_ERROR,

                            pycurl.E_FILE_COULDNT_READ_FILE):

                                # E_HTTP_RETURNED_ERROR is only used for http://

                                # and https://, but a more specific reason for

                                # failure can be obtained from respcode.

                                #

                                # E_FILE_COULDNT_READ_FILE is only used for

                                # file://, but unfortunately can mean ENOENT,

                                # EPERM, etc. and libcurl doesn't differentiate

                                # or provide a respcode.

                                if proto not in response_protocols:

                                        # For protocols that don't provide a

                                        # pycurl.RESPONSE_CODE, use the

                                        # pycurl error number instead.

                                        respcode = en

                                proto_reason = None

                                if proto in tx.proto_code_map:

                                        # Look up protocol error code map

                                        # from transport exception's table.

                                        pmap = tx.proto_code_map[proto]

                                        if respcode in pmap:

                                                proto_reason = pmap[respcode]

                                ex = tx.TransportProtoError(proto, respcode,

                                    url, reason=proto_reason, repourl=urlstem,

                                    uuid=uuid)

                                repostats.record_error(decayable=ex.decayable)

                                errors_seen += 1

                        else:

                                timeout = en == pycurl.E_OPERATION_TIMEOUTED

                                ex = tx.TransportFrameworkError(en, url, em,

                                    repourl=urlstem, uuid=uuid)

                                repostats.record_error(decayable=ex.decayable,

                                    timeout=timeout)

                                errors_seen += 1

                        if ex and ex.retryable:

                                failures.append(ex) 

                        elif ex and not ex_to_raise:

                                ex_to_raise = ex

                        done_handles.append(h)

                for h in good:

                        # Get statistics for each handle.

                        repostats = self.__xport.stats[h.repourl]

                        visited_repos.add(repostats)

                        repostats.record_tx()

                        nbytes = h.getinfo(pycurl.SIZE_DOWNLOAD)

                        seconds = h.getinfo(pycurl.TOTAL_TIME)

                        conn_count = h.getinfo(pycurl.NUM_CONNECTS)

                        conn_time = h.getinfo(pycurl.CONNECT_TIME)

                        h.filetime = h.getinfo(pycurl.INFO_FILETIME)

                        url = h.url

                        uuid = h.uuid

                        urlstem = h.repourl

                        proto = urlparse.urlsplit(url)[0]

                        # When using pipelined operations, libcurl tracks the

                        # amount of time taken for the entire pipelined request

                        # as opposed to just the amount of time for a single

                        # file in the pipeline.  So, if the connection time is 0

                        # for a request using http(s), then it was pipelined and

                        # the total time must be obtained by subtracting the

                        # time the transfer of the individual request started

                        # from the total time.

                        if conn_time == 0 and proto in pipelined_protocols:

                                # Only performing this subtraction when the

                                # conn_time is 0 allows the first request in

                                # the pipeline to properly include connection

                                # time, etc. to initiate the transfer and

                                # the correct calculations of bytespersec.

                                seconds -= h.getinfo(pycurl.STARTTRANSFER_TIME)

                        elif conn_time > 0:

                                seconds -= conn_time

                        if seconds > 0:

                                bytespersec = nbytes / seconds

                        else:

                                bytespersec = 0

                        # If a request ahead of a successful request fails due

                        # to a timeout, sometimes libcurl will report impossibly

                        # large total time values.  In this case, check that the

                        # nbytes/sec exceeds our minimum threshold.  If it does

                        # not, and the total time is longer than our timeout,

                        # discard the time calculation as it is bogus.

                        if (bytespersec < 

                            global_settings.pkg_client_lowspeed_limit) and (

                            seconds > 

                            global_settings.PKG_CLIENT_LOWSPEED_TIMEOUT):

                                nbytes = 0

                                seconds = 0

                        repostats.record_progress(nbytes, seconds)

                        # Only count connections if the connection time is

                        # positive for http(s); for all other protocols,

                        # record the connection regardless.

                        if conn_count > 0 and conn_time > 0:

                                repostats.record_connection(conn_time)

                        respcode = h.getinfo(pycurl.RESPONSE_CODE)

                        if proto not in response_protocols or \

                            respcode == httplib.OK:

                                h.success = True

                                repostats.clear_consecutive_errors()

                                success.append(url)

                        else:

                                proto_reason = None

                                if proto in tx.proto_code_map:

                                        # Look up protocol error code map

                                        # from transport exception's table.

                                        pmap = tx.proto_code_map[proto]

                                        if respcode in pmap:

                                                proto_reason = pmap[respcode]

                                ex = tx.TransportProtoError(proto,

                                    respcode, url, reason=proto_reason,

                                    repourl=urlstem, uuid=uuid)

                                # If code >= 400, record this as an error.

                                # Handlers above the engine get to decide

                                # for 200/300 codes that aren't OK

                                if respcode >= 400: 

                                        repostats.record_error(

                                            decayable=ex.decayable)

                                        errors_seen += 1

                                # If code == 0, libcurl failed to read

                                # any HTTP status.  Response is almost

                                # certainly corrupted.

                                elif respcode == 0:

                                        repostats.record_error()

                                        errors_seen += 1

                                        reason = "Invalid HTTP status code " \

                                            "from server" 

                                        ex = tx.TransportProtoError(proto,

                                            url=url, reason=reason,

                                            repourl=urlstem, uuid=uuid)

                                        ex.retryable = True 

                                # Stash retryable failures, arrange

                                # to raise first fatal error after

                                # cleanup.

                                if ex.retryable:

                                        failures.append(ex)

                                elif not ex_to_raise:

                                        ex_to_raise = ex

                        done_handles.append(h)

                # Call to remove_handle must be separate from info_read()

                for h in done_handles:

                        self.__mhandle.remove_handle(h)

                        self.__teardown_handle(h)

                        self.__freehandles.append(h)

                self.__failures = failures

                self.__success = success

                if ex_to_raise:

                        raise ex_to_raise

                # Don't bother to check the transient error count if no errors

                # were encountered in this transaction.

                if errors_seen == 0:

                        return

                # If errors were encountered, but no exception raised,

                # check if the maximum number of transient failures has

                # been exceeded at any of the endpoints that were visited

                # during this transaction.

                for rs in visited_repos:

                        numce = rs.consecutive_errors

                        if numce >= \

                            global_settings.PKG_CLIENT_MAX_CONSECUTIVE_ERROR:

                                # Reset consecutive error count before raising

                                # this exception.

                                rs.clear_consecutive_errors()

                                raise tx.ExcessiveTransientFailure(rs.url,

                                    numce)

        def check_status(self, urllist=None, good_reqs=False):

                """Return information about retryable failures that occured

                during the request.

                This is a list of transport exceptions.  Caller

                may raise these, or process them for failure information.

                Urllist is an optional argument to return only failures

                for a specific URLs.  Not all callers of check status

                want to claim the error state of all pending transactions.

                Transient errors are part of standard control flow.

                The caller will look at these and decide whether

                to throw them or not.  Permanent failures are raised

                by the transport engine as soon as they occur.

                If good_reqs is set to true, then check_stats will

                return a tuple of lists, the first list contains the

                transient errors that were encountered, the second list

                contains successfully transferred urls.  Because the

                list of successfully transferred URLs may be long,

                it is discarded if not requested by the caller."""

                # if list not specified, return all failures

                if not urllist:

                        rf = self.__failures

                        rs = self.__success

                        self.__failures = []

                        self.__success = []

                        if good_reqs:

                                return rf, rs

                        return rf

                # otherwise, look for failures that match just the URLs

                # in urllist.

                rf = [

                    tf

                    for tf in self.__failures

                    if hasattr(tf, "url") and tf.url in urllist

                ]

                # remove failues in separate pass, or else for loop gets

                # confused.

                for f in rf:

                        self.__failures.remove(f)

                if not good_reqs:

                        self.__success = []

                        return rf

                rs = []

                for ts in self.__success:

                        if ts in urllist:

                                rs.append(ts)

                for s in rs:

                        self.__success.remove(s)

                return rf, rs