Compare ⋅ cms-dev/cms

Flags

Flags have been temporarily removed from this view while the flagging feature is refactored for better performance and user experience.

You can still use flags when viewing individual files. Flag-level thresholds will also remain on pull and merge requests in your repository provider.

More information can be found in our documentation.

Showing 3 of 4 files from the diff.

74.13% 92.00% -1.52%

cms/service/EvaluationService.py changed.

86.87% 80.00% +2.33%

cms/io/triggeredservice.py changed.

82.64% 100.00% +0.95%

cms/service/esoperations.py changed.

Other files ignored by Codecov

cms/server/admin/static/aws_utils.js has changed.

@@ -40,6 +40,7 @@

Loading

from sqlalchemy.exc import IntegrityError

from cms import ServiceCoord, get_service_shards
from cmscommon.datetime import make_timestamp
from cms.db import SessionGen, Digest, Dataset, Evaluation, Submission, \
    SubmissionResult, Testcase, UserTest, UserTestResult, get_submissions, \
    get_submission_results, get_datasets_to_judge

@@ -79,6 +80,13 @@

Loading

        # Lock used to guard the currently executing operations
        self._current_execution_lock = gevent.lock.RLock()

        # As evaluate operations are split by testcases, there are too
        # many entries in the queue to display, so we just take only one
        # operation of each (type, object_id, dataset_id, priority) tuple.
        # This dictionary maps any such tuple to a "queue entry" (lacking
        # the testcase codename) and keeps track of multiplicity.
        self.queue_status_cumulative = dict()

        for i in range(get_service_shards("Worker")):
            worker = ServiceCoord("Worker", i)
            self.pool.add_worker(worker)

@@ -142,6 +150,21 @@

Loading

                    self._currently_executing = []
                    break

    def enqueue(self, item, priority, timestamp):
        success = super().enqueue(item, priority, timestamp)
        if success:
            # Add the item to the cumulative status dictionary.
            key = item.short_key() + (str(priority),)
            if key in self.queue_status_cumulative:
                self.queue_status_cumulative[key]["item"]["multiplicity"] += 1
            else:
                item_entry = item.to_dict()
                del item_entry["testcase_codename"]
                item_entry["multiplicity"] = 1
                entry = {"item": item_entry, "priority": priority, "timestamp": make_timestamp(timestamp)}
                self.queue_status_cumulative[key] = entry
        return success

    def dequeue(self, operation):
        """Remove an item from the queue.


@@ -152,7 +175,8 @@

Loading


        """
        try:
            super().dequeue(operation)
            queue_entry = super().dequeue(operation)
            self.remove_from_cumulative_status(queue_entry)
        except KeyError:
            with self._current_execution_lock:
                for i in range(len(self._currently_executing)):

@@ -161,6 +185,18 @@

Loading

                        return
            raise

    def _pop(self, wait=False):
        queue_entry = super()._pop(wait=wait)
        self.remove_from_cumulative_status(queue_entry)
        return queue_entry

    def remove_from_cumulative_status(self, queue_entry):
        # Remove the item from the cumulative status dictionary.
        key = queue_entry.item.short_key() + (str(queue_entry.priority),)
        self.queue_status_cumulative[key]["item"]["multiplicity"] -= 1
        if self.queue_status_cumulative[key]["item"]["multiplicity"] == 0:
            del self.queue_status_cumulative[key]


def with_post_finish_lock(func):
    """Decorator for locking on self.post_finish_lock.

@@ -989,30 +1025,17 @@

Loading

        the first queue.

        As evaluate operations are split by testcases, there are too
        many entries in the queue to display, so we just take only one
        operation of each (type, object_id, dataset_id)
        tuple. Generally, we will see only one evaluate operation for
        each submission in the queue status with the number of
        testcase which will be evaluated next. Moreover, we pass also
        the number of testcases in the queue.
        many entries in the queue to display, so we collect entries with the
        same (type, object_id, dataset_id, priority) tuple.
        Generally, we will see only one evaluate operation for each submission
        in the queue status.

        The entries are then ordered by priority and timestamp (the
        same criteria used to look at what to complete next).

        return ([QueueEntry]): the list with the queued elements.

        """
        entries = super().queue_status()[0]
        entries_by_key = dict()
        for entry in entries:
            key = (str(entry["item"]["type"]),
                   str(entry["item"]["object_id"]),
                   str(entry["item"]["dataset_id"]))
            if key in entries_by_key:
                entries_by_key[key]["item"]["multiplicity"] += 1
            else:
                entries_by_key[key] = entry
                entries_by_key[key]["item"]["multiplicity"] = 1
        return sorted(
            entries_by_key.values(),
            self.get_executor().queue_status_cumulative.values(),
            key=lambda x: (x["priority"], x["timestamp"]))

@@ -99,8 +99,22 @@

Loading


        item (QueueItem): the item to remove.

        return (QueueEntry): the corresponding queue entry.

        """
        return self._operation_queue.remove(item)

    def _pop(self, wait=False):
        """Extract (and return) the first element in the queue.

        wait (bool): if True, block until an element is present.

        return (QueueEntry): first element in the queue.

        raise (LookupError): on empty queue, if wait was false.

        """
        self._operation_queue.remove(item)
        return self._operation_queue.pop(wait=wait)

    def run(self):
        """Monitor the queue, and dispatch operations when available.

@@ -114,13 +128,13 @@

Loading

        """
        while True:
            # Wait for the queue to be non-empty.
            to_execute = [self._operation_queue.pop(wait=True)]
            to_execute = [self._pop(wait=True)]
            if self._batch_executions:
                max_operations = self.max_operations_per_batch()
                while not self._operation_queue.empty() and (
                        max_operations == 0 or
                        len(to_execute) < max_operations):
                    to_execute.append(self._operation_queue.pop())
                    to_execute.append(self._pop())

            assert len(to_execute) > 0, "Expected at least one element."
            if self._batch_executions:

@@ -557,3 +557,12 @@

Loading

557	557		"dataset_id": self.dataset_id,
558	558		"testcase_codename": self.testcase_codename
559	559		}
	560	+
	561	+	def short_key(self):
	562	+	"""Return a short tuple (type, object_id, dataset_id) that omits
	563	+	the testcase codename.
	564	+
	565	+	"""
	566	+	return (str(self.type_),
	567	+	str(self.object_id),
	568	+	str(self.dataset_id))

Learn more Showing 19 files with coverage changes found.

Changes in cms/io/service.py

-4

→

+4

Loading file...

Changes in cms/service/EvaluationService.py

-8

→

+8

Loading file...

Changes in cms/server/contest/handlers/tasksubmission.py

-3

→

+3

Loading file...

Changes in cms/io/priorityqueue.py

-2

→

+2

Loading file...

Changes in cms/grading/scoring.py

-1

→

+1

Loading file...

Changes in cms/db/base.py

-1

→

+1

Loading file...

Changes in cms/grading/Job.py

-2

→

+2

Loading file...

Changes in cms/io/rpc.py

New

Loading file...

Changes in cms/grading/Sandbox.py

-1

→

+1

Loading file...

Changes in cms/db/filecacher.py

-1

→

+1

Loading file...

Changes in cms/service/esoperations.py

-1

→

+1

Loading file...

Changes in cms/service/ResourceService.py

-2

→

+2

Loading file...

Changes in cms/server/admin/handlers/base.py

-4

→

+4

Loading file...

Changes in cms/db/fsobject.py

-2

→

+2

Loading file...

Changes in cms/service/ScoringService.py

-1

→

+1

Loading file...

Changes in cms/io/triggeredservice.py

-2

→

+2

Loading file...

Changes in cms/service/ProxyService.py

-6

→

+6

Loading file...

Changes in cms/service/workerpool.py

-7

→

+7

Loading file...

Changes in cms/db/util.py

-6

→

+6

Loading file...

63.70% ø +0.02%

Update some comments

fagu

#1170 67d25be

CI Passed

Hiding 2 contexual commits

+3

-3

63.68% ø +0.17%

Update description of EvaluationService.queue_status

fagu

#1170 bc1cda5

CI Passed

Hiding 3 contexual commits

+4

+31

-27

63.52% 95.83% -0.08%

Optimize queue_status on the admin interface

Unknown

#1170 9a7ac1d

CI Passed

Hiding 1 contexual commits

+16

-4

+20

Pull Request Base Commit

63.60% 0.00% +0.06%

Let PdfFileMerger open the files to be merged itself

magula

c5935e0

CI Passed

Files		•	•	•	Coverage
cms	+20	+30	ø	-10	^0.14% 73.95%
cmscommon	ø	ø	ø	ø	61.03%
cmscontrib	ø	ø	ø	ø	45.86%
cmsranking	ø	ø	ø	ø	60.73%
cmstaskenv	ø	ø	ø	ø	0.00%
Project Totals (232 files)	17,025	10,845	0	6,180	63.70%

152	175
153	176		"""
154	177		try:
155		-	super().dequeue(operation)
	178	+	queue_entry = super().dequeue(operation)
	179	+	self.remove_from_cumulative_status(queue_entry)
156	180		except KeyError:
157	181		with self._current_execution_lock:
158	182		for i in range(len(self._currently_executing)):

161	185		return
162	186		raise
163	187
	188	+	def _pop(self, wait=False):
	189	+	queue_entry = super()._pop(wait=wait)
	190	+	self.remove_from_cumulative_status(queue_entry)
	191	+	return queue_entry
	192	+
	193	+	def remove_from_cumulative_status(self, queue_entry):
	194	+	# Remove the item from the cumulative status dictionary.
	195	+	key = queue_entry.item.short_key() + (str(queue_entry.priority),)
	196	+	self.queue_status_cumulative[key]["item"]["multiplicity"] -= 1
	197	+	if self.queue_status_cumulative[key]["item"]["multiplicity"] == 0:
	198	+	del self.queue_status_cumulative[key]
	199	+
164	200
165	201		def with_post_finish_lock(func):
166	202		"""Decorator for locking on self.post_finish_lock.

989	1025		the first queue.
990	1026
991	1027		As evaluate operations are split by testcases, there are too
992		-	many entries in the queue to display, so we just take only one
993		-	operation of each (type, object_id, dataset_id)
994		-	tuple. Generally, we will see only one evaluate operation for
995		-	each submission in the queue status with the number of
996		-	testcase which will be evaluated next. Moreover, we pass also
997		-	the number of testcases in the queue.
	1028	+	many entries in the queue to display, so we collect entries with the
	1029	+	same (type, object_id, dataset_id, priority) tuple.
	1030	+	Generally, we will see only one evaluate operation for each submission
	1031	+	in the queue status.
998	1032
999	1033		The entries are then ordered by priority and timestamp (the
1000	1034		same criteria used to look at what to complete next).
1001	1035
1002	1036		return ([QueueEntry]): the list with the queued elements.
1003	1037
1004	1038		"""
1005		-	entries = super().queue_status()[0]
1006		-	entries_by_key = dict()
1007		-	for entry in entries:
1008		-	key = (str(entry["item"]["type"]),
1009		-	str(entry["item"]["object_id"]),
1010		-	str(entry["item"]["dataset_id"]))
1011		-	if key in entries_by_key:
1012		-	entries_by_key[key]["item"]["multiplicity"] += 1
1013		-	else:
1014		-	entries_by_key[key] = entry
1015		-	entries_by_key[key]["item"]["multiplicity"] = 1
1016	1039		return sorted(
1017		-	entries_by_key.values(),
	1040	+	self.get_executor().queue_status_cumulative.values(),
1018	1041		key=lambda x: (x["priority"], x["timestamp"]))

114	128		"""
115	129		while True:
116	130		# Wait for the queue to be non-empty.
117		-	to_execute = [self._operation_queue.pop(wait=True)]
	131	+	to_execute = [self._pop(wait=True)]
118	132		if self._batch_executions:
119	133		max_operations = self.max_operations_per_batch()
120	134		while not self._operation_queue.empty() and (
121	135		max_operations == 0 or
122	136		len(to_execute) < max_operations):
123		-	to_execute.append(self._operation_queue.pop())
	137	+	to_execute.append(self._pop())
124	138
125	139		assert len(to_execute) > 0, "Expected at least one element."
126	140		if self._batch_executions:

Navigation	Overlay
`t` Navigate files	`h` Toggle hits
`y` Change url to tip of branch	`m` Toggle misses
`b / v` Jump to prev/next hit line	`p` Toggle partial
`z / x` Jump to prev/next missed or partial line	`1..9` Toggle flags
`shift + o` Open current page in GitHub	`a` Toggle all on
`/ or ?` Show keyboard shortcuts dialog	`c` Toggle context lines or commits

#1170 Optimize queue_status in AWS

Flags

Show failed CI and contexual commits `c`

9 Commits

40	40		from sqlalchemy.exc import IntegrityError
41	41
42	42		from cms import ServiceCoord, get_service_shards
	43	+	from cmscommon.datetime import make_timestamp
43	44		from cms.db import SessionGen, Digest, Dataset, Evaluation, Submission, \
44	45		SubmissionResult, Testcase, UserTest, UserTestResult, get_submissions, \
45	46		get_submission_results, get_datasets_to_judge

79	80		# Lock used to guard the currently executing operations
80	81		self._current_execution_lock = gevent.lock.RLock()
81	82
	83	+	# As evaluate operations are split by testcases, there are too
	84	+	# many entries in the queue to display, so we just take only one
	85	+	# operation of each (type, object_id, dataset_id, priority) tuple.
	86	+	# This dictionary maps any such tuple to a "queue entry" (lacking
	87	+	# the testcase codename) and keeps track of multiplicity.
	88	+	self.queue_status_cumulative = dict()
	89	+
82	90		for i in range(get_service_shards("Worker")):
83	91		worker = ServiceCoord("Worker", i)
84	92		self.pool.add_worker(worker)

142	150		self._currently_executing = []
143	151		break
144	152
	153	+	def enqueue(self, item, priority, timestamp):
	154	+	success = super().enqueue(item, priority, timestamp)
	155	+	if success:
	156	+	# Add the item to the cumulative status dictionary.
	157	+	key = item.short_key() + (str(priority),)
	158	+	if key in self.queue_status_cumulative:
	159	+	self.queue_status_cumulative[key]["item"]["multiplicity"] += 1
	160	+	else:
	161	+	item_entry = item.to_dict()
	162	+	del item_entry["testcase_codename"]
	163	+	item_entry["multiplicity"] = 1
	164	+	entry = {"item": item_entry, "priority": priority, "timestamp": make_timestamp(timestamp)}
	165	+	self.queue_status_cumulative[key] = entry
	166	+	return success
	167	+
145	168		def dequeue(self, operation):
146	169		"""Remove an item from the queue.
147	170

99	99
100	100		item (QueueItem): the item to remove.
101	101
	102	+	return (QueueEntry): the corresponding queue entry.
	103	+
	104	+	"""
	105	+	return self._operation_queue.remove(item)
	106	+
	107	+	def _pop(self, wait=False):
	108	+	"""Extract (and return) the first element in the queue.
	109	+
	110	+	wait (bool): if True, block until an element is present.
	111	+
	112	+	return (QueueEntry): first element in the queue.
	113	+
	114	+	raise (LookupError): on empty queue, if wait was false.
	115	+
102	116		"""
103		-	self._operation_queue.remove(item)
	117	+	return self._operation_queue.pop(wait=wait)
104	118
105	119		def run(self):
106	120		"""Monitor the queue, and dispatch operations when available.

#1170 Optimize queue_status in AWS

Flags

Show failed CI and contexual commits c 9 Commits

Show failed CI and contexual commits `c`

9 Commits