# Copyright 2015 The Chromium OS Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
import logging
import multiprocessing
import Queue
import struct
import time
import common
from autotest_lib.client.bin import utils
from autotest_lib.client.cros.cellular.mbim_compliance \
import mbim_channel_endpoint
from autotest_lib.client.cros.cellular.mbim_compliance import mbim_errors
class MBIMChannel(object):
"""
Provide synchronous access to the modem with MBIM command level interaction.
This object should simplify your interaction over the MBIM channel as
follows:
- Use |bidirectional_transaction| to send MBIM packets that are part of a
transaction. This function will block until the transaction completes and
return the MBIM packets received in response.
- |bidirectional_transaction| will filter out packets that do not correspond
to your transaction. This way, you don't have to worry about unsolicited
notifications and/or stale packets when interacting with the modem.
- All filtered out packets can be grabbed using the
|get_outstanding_packets| function. Use this function to receive error
notifications, status notifications, etc.
- Use |unidirectional_transaction| to send MBIM packets for which you don't
expect a response.
- Use |flush| to clean out all pipes before starting a new transaction.
Note that "MBIM packets" here really means MBIM fragments. This object does
not (de)fragment packets for you. Out of necessity, it does check that
received fragments are contiguous and in-order.
So, this object houses the minimum information necessary about the MBIM
fragments to provide you a comfortable synchronous packet level channel.
"""
ENDPOINT_JOIN_TIMEOUT_S = 5
FRAGMENT_TIMEOUT_S = 3
# TODO(pprabhu) Consider allowing each transaction to specify its own
# timeout.
TRANSACTION_TIMEOUT_S = 5
MESSAGE_HEADER_FORMAT = '<LLL'
FRAGMENT_HEADER_FORMAT = '<LL'
MBIM_FRAGMENTED_MESSAGES = [
0x00000003, # MBIM_COMMAND_MSG
0x80000003, # MBIM_COMMAND_DONE
0x80000007] # MBIM_INDICATE_STATUS
def __init__(self,
device,
interface_number,
interrupt_endpoint_address,
in_buffer_size,
process_class=None):
"""
@param device: Device handle returned by PyUSB for the modem to test.
@param interface_number: |bInterfaceNumber| of the MBIM interface.
@param interrupt_endpoint_address: |bEndpointAddress| for the usb
INTERRUPT IN endpoint for notifications.
@param in_buffer_size: The (fixed) buffer size to use for in control
transfers.
@param process_class: The class to instantiate to create a subprocess.
This is used by tests only, to easily mock out the process
ceation.
"""
self._stop_request_event = multiprocessing.Event()
self._request_queue = multiprocessing.Queue()
self._response_queue = multiprocessing.Queue()
self._outstanding_packets = []
self._last_response = []
self._stashed_first_fragment = None
if process_class is None:
process_class = multiprocessing.Process
self._endpoint_process = process_class(
target=mbim_channel_endpoint.MBIMChannelEndpoint,
args=(device,
interface_number,
interrupt_endpoint_address,
in_buffer_size,
self._request_queue,
self._response_queue,
self._stop_request_event))
self._endpoint_process.start()
def __del__(self):
"""
The destructor.
Note that it is not guaranteed that |__del__| is called for objects that
exist when the interpreter exits. It is recommended to call |close|
explicitly.
"""
self.close()
def close(self):
"""
Cleanly close the MBIMChannel.
MBIMChannel forks a subprocess to communicate with the USB device. It is
recommended that |close| be called explicitly.
"""
if not self._endpoint_process:
return
if self._endpoint_process.is_alive():
self._stop_request_event.set()
self._endpoint_process.join(self.ENDPOINT_JOIN_TIMEOUT_S)
if self._endpoint_process.is_alive():
self._endpoint_process.terminate()
self._endpoint_process = None
def bidirectional_transaction(self, *args):
"""
Execute a synchronous bidirectional transaction.
@param *args: Fragments of a single MBIM transaction. An MBIM
transaction may consist of multiple fragments - each fragment is
the payload for a USB control message. It should be an
|array.array| object. It is your responsibility (and choice) to
keep the fragments in-order, and to send all the fragments.
For more details, see "Fragmentation of messages" in the MBIM
spec.
@returns: A list of fragments in the same order as received that
correspond to the given transaction. If we receive less
fragments than claimed, we will return what we get. If we
receive non-contiguous / out-of-order fragments, we'll complain.
@raises: MBIMComplianceChannelError if received fragments are
out-of-order or non-contigouos.
"""
self._verify_endpoint_open()
if not args:
mbim_errors.log_and_raise(
mbim_errors.MBIMComplianceChannelError,
'No data given to |bidirectional_transaction|.')
transaction_id, _, _ = self._fragment_metadata(args[0])
for fragment in args:
self._request_queue.put_nowait(fragment)
return self._get_response_fragments(transaction_id)
def unidirectional_transaction(self, *args):
"""
Execute a synchronous unidirectional transaction. No return value.
@param *args: Fragments of a single MBIM transaction. An MBIM
transaction may consist of multiple fragments - each fragment is
the payload for a USB control message. It should be an
|array.array| object. It is your responsibility (and choice) to
keep the fragments in-order, and to send all the fragments.
For more details, see "Fragmentation of messages" in the MBIM
spec.
"""
self._verify_endpoint_open()
if not args:
mbim_errors.log_and_raise(
mbim_errors.MBIMComplianceChannelError,
'No data given to |unidirectional_transaction|.')
for fragment in args:
self._request_queue.put_nowait(fragment)
def flush(self):
"""
Clean out all queues.
This waits till all outgoing packets have been sent, and then waits some
more to give the channel time to settle down.
@raises: MBIMComplianceChannelError if things don't settle down fast
enough.
"""
self._verify_endpoint_open()
num_remaining_fragments = self._request_queue.qsize()
try:
timeout = self.FRAGMENT_TIMEOUT_S * num_remaining_fragments
utils.poll_for_condition(lambda: self._request_queue.empty(),
timeout=timeout)
except utils.TimeoutError:
mbim_errors.log_and_raise(
mbim_errors.MBIMComplianceChannelError,
'Could not flush request queue.')
# Now wait for the response queue to settle down.
# In the worst case, each request fragment that was remaining at the
# time flush was called belonged to a different transaction, and each of
# these transactions would serially timeout in |TRANSACTION_TIMEOUT_S|.
# To avoid sleeping for long times, we cap this value arbitrarily to 5
# transactions.
num_remaining_transactions = min(5, num_remaining_fragments)
time.sleep(num_remaining_fragments * self.TRANSACTION_TIMEOUT_S)
extra_packets = self.get_outstanding_packets()
for packet in extra_packets:
logging.debug('flush: discarding packet: %s', packet)
def get_outstanding_packets(self):
"""
Get all received packets that were not part of an explicit transaction.
@returns: A list of packets. Each packet is a list of fragments, so you
perhaps want to do something like:
for packet in channel.get_outstanding_packets():
for fragment in packet:
# handle fragment.
"""
self._verify_endpoint_open()
# Try to get more packets from the response queue.
# This can block forever if the modem keeps spewing trash at us.
while True:
packet = self._get_packet_fragments()
if not packet:
break
self._outstanding_packets.append(packet)
packets = self._outstanding_packets
self._outstanding_packets = []
return packets
def _get_response_fragments(self, transaction_id):
"""
Get response for the given |transaction_id|.
@returns: A list of fragments.
@raises: MBIMComplianceChannelError if response is not recieved.
"""
def _poll_response():
packet = self._get_packet_fragments()
if not packet:
return False
first_fragment = packet[0]
response_id, _, _ = self._fragment_metadata(first_fragment)
if response_id == transaction_id:
self._last_response = packet
return True
self._outstanding_packets.append(packet)
return False
try:
utils.poll_for_condition(
_poll_response,
timeout=self.TRANSACTION_TIMEOUT_S)
except utils.TimeoutError:
mbim_errors.log_and_raise(
mbim_errors.MBIMComplianceChannelError,
'Did not receive timely reply to transaction %d' %
transaction_id)
return self._last_response
def _get_packet_fragments(self):
"""
Get all fragements of the next packet from the modem.
This function is responsible for putting together fragments of one
packet, and checking that fragments are continguous and in-order.
"""
fragments = []
if self._stashed_first_fragment is not None:
first_fragment = self._stashed_first_fragment
self._stashed_first_fragment = None
else:
try:
first_fragment = self._response_queue.get(
True, self.FRAGMENT_TIMEOUT_S)
except Queue.Empty:
# *Don't fail* Just return nothing.
return fragments
transaction_id, total_fragments, current_fragment = (
self._fragment_metadata(first_fragment))
if current_fragment != 0:
mbim_errors.log_and_raise(
mbim_errors.MBIMComplianceChannelError,
'First fragment reports fragment number %d' %
current_fragment)
fragments.append(first_fragment)
last_fragment = 0
while last_fragment < total_fragments - 1:
try:
fragment = self._response_queue.get(True,
self.FRAGMENT_TIMEOUT_S)
except Queue.Empty:
# *Don't fail* Just return the fragments we got so far.
break
fragment_id, fragment_total, fragment_current = (
self._fragment_metadata(fragment))
if fragment_id != transaction_id:
# *Don't fail* Treat a different transaction id as indicating
# that the next packet has already arrived.
logging.warning('Recieved only %d out of %d fragments for '
'transaction %d.',
last_fragment,
total_fragments,
transaction_id)
self._stashed_first_fragment = fragment
break
if fragment_total != total_fragments:
mbim_errors.log_and_raise(
mbim_errors.MBIMComplianceChannelError,
'Fragment number %d reports incorrect total (%d/%d)' %
(last_fragment + 1, fragment_total, total_fragments))
if fragment_current != last_fragment + 1:
mbim_errors.log_and_raise(
mbim_errors.MBIMComplianceChannelError,
'Received reordered fragments. Expected %d, got %d' %
(last_fragment + 1, fragment_current))
last_fragment += 1
fragments.append(fragment)
return fragments
def _fragment_metadata(self, fragment):
""" This function houses all the MBIM packet knowledge. """
# All packets have a message header.
if len(fragment) < struct.calcsize(self.MESSAGE_HEADER_FORMAT):
mbim_errors.log_and_raise(
mbim_errors.MBIMComplianceChannelError,
'Corrupted fragment |%s| does not have an MBIM header.' %
fragment)
message_type, _, transaction_id = struct.unpack_from(
self.MESSAGE_HEADER_FORMAT,
fragment)
if message_type in self.MBIM_FRAGMENTED_MESSAGES:
fragment = fragment[struct.calcsize(self.MESSAGE_HEADER_FORMAT):]
if len(fragment) < struct.calcsize(self.FRAGMENT_HEADER_FORMAT):
mbim_errors.log_and_raise(
mbim_errors.MBIMComplianceChannelError,
'Corrupted fragment |%s| does not have a fragment '
'header. ' %
fragment)
total_fragments, current_fragment = struct.unpack_from(
self.FRAGMENT_HEADER_FORMAT,
fragment)
else:
# For other types, there is only one 'fragment'.
total_fragments = 1
current_fragment = 0
return transaction_id, total_fragments, current_fragment
def _verify_endpoint_open(self):
if not self._endpoint_process.is_alive():
mbim_errors.log_and_raise(
mbim_errors.MBIMComplianceChannelError,
'MBIMChannelEndpoint died unexpectedly. '
'The actual exception can be found in log entries from the '
'subprocess.')