def perform_processing(self, add_events_request, current_time=None):
        if current_time is None:
            current_time = time.time()

        # If this is our first time processing this log file, just pretend like we had a recent success.
        if self.__last_success is None:
            self.__last_success = current_time

        self.__lock.acquire()
        self.__last_scan_time = current_time
        self.__lock.release()

        self.__log_file_iterator.scan_for_new_bytes(current_time=current_time)

        # Check to see if we haven't had a success in enough time.  If so, then we just skip ahead.
        if current_time - self.__last_success > self.__copy_staleness_threshold:
            self.skip_to_end('Too long since last success.  Last success was \'%s\'' % scalyr_util.format_time(
                self.__last_success), 'skipForStaleness', current_time=current_time)
        # Also make sure we are at least within 5MB of the tail of the log.  If not, then we skip ahead.
        elif self.__log_file_iterator.available > self.__max_log_offset_size:
            self.skip_to_end(
                'Too far behind end of log.  Num of bytes to end is %ld' % self.__log_file_iterator.available,
                'skipForTooFarBehind', current_time=current_time)

        # Keep track of both the position in the iterator and where we are about to add new events to the request,
        # in case we have to roll it back.
        original_position = self.__log_file_iterator.tell()
        original_events_position = add_events_request.position()

        start_process_time = 0.0

        # Some performance analysis timings we use from time to time.
        # fast_get_time = timeit.default_timer
        # start_process_time = fast_get_time()

        position = self.__log_file_iterator.tell()

        # noinspection PyBroadException
        try:
            # Keep track of some states about the lines/events we process.
            bytes_read = 0L
            lines_read = 0L
            bytes_copied = 0L
            lines_copied = 0L
            total_redactions = 0L
            lines_dropped_by_sampling = 0L
            bytes_dropped_by_sampling = 0L

            time_spent_reading = 0.0
            time_spent_serializing = 0.0

            buffer_filled = False
            added_thread_id = False

            # Keep looping, add more events until there are no more or there is no more room.
            while True:
                position = self.__log_file_iterator.tell(dest=position)

                #time_spent_reading -= fast_get_time()
                line = self.__log_file_iterator.readline(current_time=current_time)
                line_len = len(line)
                #time_spent_reading += fast_get_time()

                # This means we hit the end of the file, or at least there is not a new line yet available.
                if line_len == 0:
                    break

                # We have a line, process it and see what comes out.
                bytes_read += line_len
                lines_read += 1L

                if self.__num_redaction_and_sampling_rules > 0:
                    sample_result = self.__sampler.process_line(line)
                    if sample_result is None:
                        lines_dropped_by_sampling += 1L
                        bytes_dropped_by_sampling += line_len
                        continue

                    (line, redacted) = self.__redacter.process_line(line)
                    line_len = len(line)
                else:
                    sample_result = 1.0
                    redacted = False

                if line_len > 0:
                    # Try to add the line to the request, but it will let us know if it exceeds the limit it can
                    # send.
                    sequence_id, sequence_number = self.__log_file_iterator.get_sequence()

                    #time_spent_serializing += fast_get_time()
                    event = self.__create_events_object(line, sample_result)
                    if not add_events_request.add_event(event, sequence_id=sequence_id, sequence_number=sequence_number):
                        #time_spent_serializing -= fast_get_time()

                        self.__log_file_iterator.seek(position)
                        buffer_filled = True
                        break

                    #time_spent_serializing -= fast_get_time()

                    # Try to add the thread id if we have not done so far.  This should only be added once per file.
                    if not added_thread_id:
                        if not add_events_request.add_thread(self.__thread_id, self.__thread_name):
                            # If we got here, it means we did not have enough room to add both the thread id
                            # and the event into the events request.  So, we have to remove the event we just
                            # added to the add_events_request by setting the position to the original.
                            add_events_request.set_position(original_events_position)
                            self.__log_file_iterator.seek(position)
                            buffer_filled = True
                            break
                        added_thread_id = True

                if redacted:
                    total_redactions += 1L
                bytes_copied += line_len
                lines_copied += 1

            final_position = self.__log_file_iterator.tell()

            # start_process_time = fast_get_time() - start_process_time
            #add_events_request.increment_timing_data(serialization_time=time_spent_serializing,
            #                                         reading_time=time_spent_reading,
            #                                         process_time=start_process_time,
            #                                         lines=lines_read, bytes=bytes_read, files=1)
            add_events_request.increment_timing_data(lines=lines_read, bytes=bytes_read, files=1)

            # To do proper account when an RPC has failed and we retry it, we track how many bytes are
            # actively being processed.  We will update this once the completion callback has been invoked.
            self.__lock.acquire()
            self.__total_bytes_being_processed = bytes_copied
            self.__total_bytes_pending = self.__log_file_iterator.available
            self.__lock.release()

            # We have finished a processing loop.  We probably won't be calling the iterator for a while, so let it
            # do some clean up work until the next time we need it.
            self.__log_file_iterator.prepare_for_inactivity()

            # If we get here on what was a new file, then the file is now an existing file, so set the maximum log
            # offset size to use the size for existing files
            self.__max_log_offset_size = self.__max_existing_log_offset_size

            # Define the callback to return.
            def completion_callback(result):
                """Invoked by the caller to indicate if the events were successfully sent to server, and if not,
                what to do.

                @param result: Must be one of SUCCESS, FAIL_AND_DROP, FAIL_AND_RETRY.
                @type result: int
                @return: True if the processor has been closed because it is finished and the caller should no longer
                    use it.
                @rtype: bool
                """
                try:
                    log.log(scalyr_logging.DEBUG_LEVEL_3, 'Result for advancing %s was %s', self.__path, str(result))
                    self.__lock.acquire()
                    # Zero out the bytes we were tracking as they were in flight.
                    self.__total_bytes_being_processed = 0

                    # If it was a success, then we update the counters and advance the iterator.
                    if result == LogFileProcessor.SUCCESS:
                        self.__total_bytes_copied += bytes_copied
                        self.__total_bytes_skipped += self.__log_file_iterator.bytes_between_positions(
                            original_position, final_position) - bytes_read

                        self.__total_bytes_dropped_by_sampling += bytes_dropped_by_sampling
                        self.__total_bytes_pending = self.__log_file_iterator.available
                        self.__total_lines_copied += lines_copied
                        self.__total_lines_dropped_by_sampling += lines_dropped_by_sampling
                        self.__total_redactions += total_redactions
                        self.__last_success = current_time

                        # Do a mark to cleanup any state in the iterator.  We know we won't have to roll back
                        # to before this point now.
                        self.__log_file_iterator.mark(final_position, current_time=current_time)
                        if self.__log_file_iterator.at_end:
                            self.__log_file_iterator.close()
                            self.__is_closed = True
                            return True
                        else:
                            return False
                    elif result == LogFileProcessor.FAIL_AND_DROP:
                        self.__log_file_iterator.mark(final_position, current_time=current_time)
                        self.__total_bytes_pending = self.__log_file_iterator.available
                        self.__total_bytes_failed += bytes_read
                        return False
                    elif result == LogFileProcessor.FAIL_AND_RETRY:
                        self.__log_file_iterator.seek(original_position)
                        self.__total_bytes_pending = self.__log_file_iterator.available
                        return False
                    else:
                        raise Exception('Invalid result %s' % str(result))
                finally:
                    self.__lock.release()

            log.log(scalyr_logging.DEBUG_LEVEL_3, 'Scanned %s and found %ld bytes for copying.  Buffer filled=%s.',
                    self.__path, bytes_copied, str(buffer_filled))

            return completion_callback, buffer_filled
        except Exception:
            log.exception('Failed to copy lines from \'%s\'.  Will re-attempt lines.', self.__path,
                          error_code='logCopierFailed')
            log.log(scalyr_logging.DEBUG_LEVEL_3, 'Failed while scanning \'%s\' for new bytes.', self.__path)

            # Roll back the positions if something happened.
            self.__log_file_iterator.seek(original_position)
            add_events_request.set_position(original_events_position)

            return None, False
Expand ST build system and OS environment variables to normalized path
dataset/ETHPy150Open scalyr/scalyr-agent-2/scalyr_agent/log_processing.py/LogFileProcessor.perform_processing--crossover with--dataset/ETHPy150Open srusskih/SublimeJEDI/sublime_jedi/utils.py/expand_path