Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[EventHubs] Performance improvement #7533

Merged
merged 7 commits into from
Oct 2, 2019
Merged

[EventHubs] Performance improvement #7533

Show file tree
Hide file tree
Changes from 4 commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
41cce74
performance improvement
yunhaoling Sep 29, 2019
25fa5d7
batch offset
yunhaoling Sep 29, 2019
24a181d
merge eventhub-preview4
yunhaoling Oct 1, 2019
647b256
Review feedback
yunhaoling Oct 2, 2019
64c7152
Remove lazy parse in EventData
yunhaoling Oct 2, 2019
3952b51
Merge branch 'eventhubs_preview4' into performance-improvement
yunhaoling Oct 2, 2019
803bc1e
Add annotation assertions
yunhaoling Oct 2, 2019
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 8 additions & 3 deletions sdk/eventhub/azure-eventhubs/azure/eventhub/aio/consumer_async.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,7 @@

from uamqp import errors, types, utils # type: ignore
from uamqp import ReceiveClientAsync, Source # type: ignore
import uamqp

from azure.eventhub import EventData, EventPosition
from azure.eventhub.error import EventHubError, ConnectError, _error_handler
Expand Down Expand Up @@ -118,7 +119,7 @@ async def __anext__(self):
self._offset = EventPosition(event_data.offset, inclusive=False)
retried_times = 0
if self._track_last_enqueued_event_properties:
self._last_enqueued_event_properties = event_data._runtime_info # pylint:disable=protected-access
self._last_enqueued_event_properties = event_data._get_last_enqueued_event_properties() # pylint:disable=protected-access
return event_data
except Exception as exception: # pylint:disable=broad-except
last_exception = await self._handle_exception(exception)
Expand Down Expand Up @@ -153,6 +154,8 @@ def _create_handler(self):
error_policy=self._retry_policy,
keep_alive_interval=self._keep_alive,
client_name=self._name,
receive_settle_mode=uamqp.constants.ReceiverSettleMode.ReceiveAndDelete,
auto_complete=False,
properties=self._client._create_properties( # pylint:disable=protected-access
self._client._config.user_agent), # pylint:disable=protected-access
desired_capabilities=desired_capabilities, # pylint:disable=protected-access
Expand Down Expand Up @@ -199,12 +202,14 @@ async def _receive(self, timeout_time=None, max_batch_size=None, **kwargs):
timeout=remaining_time_ms)
for message in message_batch:
event_data = EventData._from_message(message) # pylint:disable=protected-access
self._offset = EventPosition(event_data.offset)
data_batch.append(event_data)
event_data._trace_link_message() # pylint:disable=protected-access

if data_batch:
self._offset = EventPosition(data_batch[-1].offset)

if self._track_last_enqueued_event_properties and len(data_batch):
self._last_enqueued_event_properties = data_batch[-1]._runtime_info # pylint:disable=protected-access
self._last_enqueued_event_properties = data_batch[-1]._get_last_enqueued_event_properties() # pylint:disable=protected-access

return data_batch

Expand Down
59 changes: 43 additions & 16 deletions sdk/eventhub/azure-eventhubs/azure/eventhub/common.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -78,7 +78,8 @@ def __init__(self, body=None, to_device=None):
self._delivery_annotations = {}
self._app_properties = {}
self._msg_properties = MessageProperties()
self._runtime_info = {}
self._last_enqueued_event_properties = {}
self._need_further_parse = False
if to_device:
self._msg_properties.to = '/devices/{}/messages/devicebound'.format(to_device)
if body and isinstance(body, list):
Expand Down Expand Up @@ -152,35 +153,49 @@ def _trace_link_message(self, parent_span=None):
if traceparent:
current_span.link(traceparent)

@staticmethod
def _from_message(message):
# pylint:disable=protected-access
event_data = EventData(body='')
event_data.message = message
event_data._msg_properties = message.properties
event_data._annotations = message.annotations
event_data._app_properties = message.application_properties
event_data._delivery_annotations = message.delivery_annotations
if event_data._delivery_annotations:
event_data._runtime_info = {
def _get_last_enqueued_event_properties(self):
if self._last_enqueued_event_properties:
return self._last_enqueued_event_properties

if self.message.delivery_annotations:
self._last_enqueued_event_properties = {
"sequence_number":
event_data._delivery_annotations.get(EventData.PROP_LAST_ENQUEUED_SEQUENCE_NUMBER, None),
self.message.delivery_annotations.get(EventData.PROP_LAST_ENQUEUED_SEQUENCE_NUMBER, None),
"offset":
event_data._delivery_annotations.get(EventData.PROP_LAST_ENQUEUED_OFFSET, None),
self.message.delivery_annotations.get(EventData.PROP_LAST_ENQUEUED_OFFSET, None),
"enqueued_time":
event_data._delivery_annotations.get(EventData.PROP_LAST_ENQUEUED_TIME_UTC, None),
self.message.delivery_annotations.get(EventData.PROP_LAST_ENQUEUED_TIME_UTC, None),
"retrieval_time":
event_data._delivery_annotations.get(EventData.PROP_RUNTIME_INFO_RETRIEVAL_TIME_UTC, None)
self.message.delivery_annotations.get(EventData.PROP_RUNTIME_INFO_RETRIEVAL_TIME_UTC, None)
}
return self._last_enqueued_event_properties

return None

@staticmethod
def _from_message(message):
annatisch marked this conversation as resolved.
Show resolved Hide resolved
# pylint:disable=protected-access
event_data = EventData(body='')
event_data.message = message
event_data._need_further_parse = True
return event_data

def _parse_message_properties(self):
self._msg_properties = self.message.properties
self._annotations = self.message.annotations
self._app_properties = self.message.application_properties
self._delivery_annotations = self.message.delivery_annotations
self._need_further_parse = False

@property
def sequence_number(self):
"""
The sequence number of the event data object.

:rtype: int or long
"""
if self._need_further_parse:
self._parse_message_properties()
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

uamqp already lazily loads properties. I guess another lazy loading won't improve much performance

Copy link
Contributor Author

@yunhaoling yunhaoling Oct 2, 2019 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'm afraid I have to do the lazy assignment here.
because the parsing actually takes place when properties of message are assigned to events (the property method of message in uamqp will call parse function).
so we need make the assignment lazy too.

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think we can potentially only put the lazy loading into uamqp... we can discuss offline.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

lazy parsing removed

return self._annotations.get(EventData.PROP_SEQ_NUMBER, None)

@property
Expand All @@ -190,6 +205,8 @@ def offset(self):

:rtype: str
"""
if self._need_further_parse:
self._parse_message_properties()
try:
return self._annotations[EventData.PROP_OFFSET].decode('UTF-8')
except (KeyError, AttributeError):
Expand All @@ -202,6 +219,8 @@ def enqueued_time(self):

:rtype: datetime.datetime
"""
if self._need_further_parse:
self._parse_message_properties()
timestamp = self._annotations.get(EventData.PROP_TIMESTAMP, None)
if timestamp:
return datetime.datetime.utcfromtimestamp(float(timestamp)/1000)
Expand All @@ -215,6 +234,8 @@ def device_id(self):

:rtype: bytes
"""
if self._need_further_parse:
self._parse_message_properties()
return self._annotations.get(EventData.PROP_DEVICE_ID, None)

@property
Expand All @@ -224,6 +245,8 @@ def partition_key(self):

:rtype: bytes
"""
if self._need_further_parse:
self._parse_message_properties()
try:
return self._annotations[EventData.PROP_PARTITION_KEY_AMQP_SYMBOL]
except KeyError:
Expand All @@ -236,6 +259,8 @@ def application_properties(self):

:rtype: dict
"""
if self._need_further_parse:
self._parse_message_properties()
return self._app_properties

@application_properties.setter
Expand All @@ -257,6 +282,8 @@ def system_properties(self):

:rtype: dict
"""
if self._need_further_parse:
self._parse_message_properties()
return self._annotations

@property
Expand Down
11 changes: 8 additions & 3 deletions sdk/eventhub/azure-eventhubs/azure/eventhub/consumer.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,7 @@

from uamqp import types, errors, utils # type: ignore
from uamqp import ReceiveClient, Source # type: ignore
import uamqp

from azure.eventhub.common import EventData, EventPosition
from azure.eventhub.error import _error_handler
Expand Down Expand Up @@ -114,7 +115,7 @@ def __next__(self):
self._offset = EventPosition(event_data.offset, inclusive=False)
retried_times = 0
if self._track_last_enqueued_event_properties:
self._last_enqueued_event_properties = event_data._runtime_info # pylint:disable=protected-access
self._last_enqueued_event_properties = event_data._get_last_enqueued_event_properties() # pylint:disable=protected-access
annatisch marked this conversation as resolved.
Show resolved Hide resolved
return event_data
except Exception as exception: # pylint:disable=broad-except
last_exception = self._handle_exception(exception)
Expand Down Expand Up @@ -149,6 +150,8 @@ def _create_handler(self):
error_policy=self._retry_policy,
keep_alive_interval=self._keep_alive,
client_name=self._name,
receive_settle_mode=uamqp.constants.ReceiverSettleMode.ReceiveAndDelete,
auto_complete=False,
properties=self._client._create_properties( # pylint:disable=protected-access
self._client._config.user_agent), # pylint:disable=protected-access
desired_capabilities=desired_capabilities) # pylint:disable=protected-access
Expand Down Expand Up @@ -193,12 +196,14 @@ def _receive(self, timeout_time=None, max_batch_size=None, **kwargs):
timeout=remaining_time_ms)
for message in message_batch:
event_data = EventData._from_message(message) # pylint:disable=protected-access
self._offset = EventPosition(event_data.offset)
data_batch.append(event_data)
event_data._trace_link_message() # pylint:disable=protected-access

if data_batch:
self._offset = EventPosition(data_batch[-1].offset)

if self._track_last_enqueued_event_properties and len(data_batch):
self._last_enqueued_event_properties = data_batch[-1]._runtime_info # pylint:disable=protected-access
self._last_enqueued_event_properties = data_batch[-1]._get_last_enqueued_event_properties() # pylint:disable=protected-access

return data_batch

Expand Down