Problem Description

This problem asks you to implement a task management system that can handle multiple users and their tasks. Each task has a priority level, and the system needs to support several operations.

The TaskManager class needs to support the following operations:

1. Initialization: Create the task manager with an initial list of tasks. Each task is represented as [userId, taskId, priority].

2. Add a task: Add a new task to a specific user with add(userId, taskId, priority). The taskId is guaranteed to be unique (not already in the system).

3. Edit a task: Change the priority of an existing task with edit(taskId, newPriority). The taskId is guaranteed to exist in the system.

4. Remove a task: Delete a task from the system with rmv(taskId). The taskId is guaranteed to exist.

5. Execute top priority task: Execute and remove the highest priority task across all users with execTop(). If multiple tasks have the same highest priority, choose the one with the highest taskId. Return the userId of the executed task, or -1 if no tasks exist.

The solution uses two data structures:

• A dictionary d that maps each taskId to a tuple of (userId, priority)
• A SortedList that maintains tasks sorted by priority and taskId

The key insight is storing tasks as (-priority, -taskId) in the sorted list. This ensures that:

• Higher priorities come first (since we negate the priority)
• Among tasks with the same priority, higher taskIds come first (since we negate the taskId)

When executing the top task, the system simply pops the first element from the sorted list, retrieves the corresponding user information from the dictionary, and returns the userId.

Intuition

The core challenge is maintaining tasks in a way that allows us to quickly find and execute the highest priority task while also supporting efficient updates and deletions.

Let's think about what data we need to track:

• For each task, we need to know which user owns it and its priority
• We need to find the highest priority task quickly
• When priorities are tied, we need the highest taskId

A natural first thought might be to use a heap or priority queue. However, standard heaps don't support efficient arbitrary deletions or updates - we'd need to search through the entire heap to find a specific taskId for editing or removal.

This leads us to consider maintaining two separate data structures:

1. A mapping to quickly look up task information by taskId
2. A sorted structure to maintain priority ordering

For the mapping, a dictionary is perfect - it gives us O(1) access to any task's information using its taskId as the key. We store (userId, priority) as the value.

For maintaining sorted order, we need a structure that supports:

• Insertion in sorted order
• Removal of arbitrary elements
• Access to the maximum element

A SortedList (balanced binary search tree) fits perfectly, providing O(log n) operations for all these needs.

The clever trick is how we store elements in the sorted list. Since we want the highest priority first but Python's sorted structures maintain ascending order, we store negative priorities (-priority). Similarly, when priorities are equal, we want the highest taskId first, so we also negate the taskId (-taskId). This transforms our "find maximum" problem into a "find minimum" problem, which naturally aligns with the sorted order.

By maintaining these two synchronized data structures, we achieve efficient operations: O(log n) for add, edit, and remove operations, and O(log n) for executing the top task.

Learn more about Heap (Priority Queue) patterns.

Solution Approach

The solution uses two synchronized data structures to efficiently manage tasks:

1. Dictionary d: Maps taskId to (userId, priority) for O(1) lookups
2. SortedList st: Maintains tasks in sorted order by (-priority, -taskId)

Implementation Details:

Initialization (__init__):

def __init__(self, tasks: List[List[int]]):
    self.d = {}
    self.st = SortedList()
    for task in tasks:
        self.add(*task)

• Initialize empty dictionary and sorted list
• Use the add method to process each initial task

Adding a Task (add):

def add(self, userId: int, taskId: int, priority: int) -> None:
    self.d[taskId] = (userId, priority)
    self.st.add((-priority, -taskId))

• Store task information in dictionary with taskId as key
• Add (-priority, -taskId) tuple to sorted list to maintain proper ordering

Editing Task Priority (edit):

def edit(self, taskId: int, newPriority: int) -> None:
    userId, priority = self.d[taskId]
    self.st.discard((-priority, -taskId))
    self.d[taskId] = (userId, newPriority)
    self.st.add((-newPriority, -taskId))

• Retrieve current user and priority from dictionary
• Remove old entry from sorted list using discard (won't raise error if not found)
• Update dictionary with new priority
• Add new entry to sorted list with updated priority

Removing a Task (rmv):

def rmv(self, taskId: int) -> None:
    _, priority = self.d[taskId]
    self.d.pop(taskId)
    self.st.remove((-priority, -taskId))

• Get priority from dictionary (userId not needed)
• Remove from dictionary using pop
• Remove from sorted list using remove

Executing Top Priority Task (execTop):

def execTop(self) -> int:
    if not self.st:
        return -1
    taskId = -self.st.pop(0)[1]
    userId, _ = self.d[taskId]
    self.d.pop(taskId)
    return userId

• Check if sorted list is empty, return -1 if no tasks
• Pop the first element (highest priority due to negative values)
• Extract taskId by negating the second element of the tuple
• Get userId from dictionary
• Remove task from dictionary
• Return the userId

Time Complexity:

• add: O(log n) for sorted list insertion
• edit: O(log n) for removal and insertion in sorted list
• rmv: O(log n) for sorted list removal
• execTop: O(log n) for popping from sorted list
• Space: O(n) for storing n tasks

The negative value trick (-priority, -taskId) ensures that the sorted list maintains tasks in descending order of priority and taskId, making the first element always the correct one to execute.

Example Walkthrough

Let's walk through a small example to illustrate how the solution works.

Initial Setup:

tasks = [[1, 101, 3], [2, 102, 5], [1, 103, 3]]
tm = TaskManager(tasks)

After initialization:

• Dictionary d: {101: (1, 3), 102: (2, 5), 103: (1, 3)}
• SortedList st: [(-5, -102), (-3, -103), (-3, -101)]

Notice how the sorted list orders tasks:

• Task 102 with priority 5 comes first (stored as -5)
• Tasks 101 and 103 both have priority 3, but 103 has higher taskId, so (-3, -103) comes before (-3, -101)

Operation 1: Execute top priority task

result = tm.execTop()  # Returns 2

• The first element in sorted list is (-5, -102)
• Extract taskId: -(-102) = 102
• Look up in dictionary: d[102] = (2, 5), so userId = 2
• Remove task 102 from both structures
• After: d = {101: (1, 3), 103: (1, 3)}, st = [(-3, -103), (-3, -101)]

Operation 2: Add a new task

tm.add(3, 104, 3)

• Add to dictionary: d[104] = (3, 3)
• Add to sorted list: (-3, -104)
• After: d = {101: (1, 3), 103: (1, 3), 104: (3, 3)}, st = [(-3, -104), (-3, -103), (-3, -101)]
• Note: Task 104 is now first among priority-3 tasks due to highest taskId

Operation 3: Edit task priority

tm.edit(101, 7)

• Current info: d[101] = (1, 3)
• Remove (-3, -101) from sorted list
• Update dictionary: d[101] = (1, 7)
• Add (-7, -101) to sorted list
• After: d = {101: (1, 7), 103: (1, 3), 104: (3, 3)}, st = [(-7, -101), (-3, -104), (-3, -103)]
• Task 101 is now the highest priority task

Operation 4: Execute top priority task again

result = tm.execTop()  # Returns 1

• Pop (-7, -101) from sorted list
• taskId = 101, userId = 1
• After: d = {103: (1, 3), 104: (3, 3)}, st = [(-3, -104), (-3, -103)]

This example demonstrates how the negative value storage ensures correct priority ordering and how the two data structures work together to maintain efficient operations.

Solution Implementation

Time and Space Complexity

Time Complexity:

• __init__(tasks): O(n log n) where n is the number of tasks. Each task is added to the dictionary in O(1) and to the SortedList in O(log n), resulting in O(n log n) for all tasks.

• add(userId, taskId, priority): O(log n) where n is the current number of tasks. Dictionary insertion is O(1), but adding to SortedList requires O(log n) for maintaining sorted order.

• edit(taskId, newPriority): O(log n) where n is the current number of tasks. The operation involves:

  • Dictionary lookup: O(1)
  • Discarding from SortedList: O(log n) for finding + O(n) worst case for removal
  • Dictionary update: O(1)
  • Adding to SortedList: O(log n)
  • Overall: O(n) worst case, but typically O(log n) in practice

• rmv(taskId): O(n) worst case where n is the current number of tasks. The operation involves:

  • Dictionary lookup: O(1)
  • Dictionary deletion: O(1)
  • Removing from SortedList: O(log n) for finding + O(n) for removal
  • Overall: O(n) due to the remove() operation

• execTop(): O(log n) where n is the current number of tasks. Popping from the front of SortedList is O(log n), and dictionary operations are O(1).

Space Complexity:

• Overall space complexity: O(n) where n is the number of tasks
• Dictionary self.d: O(n) to store all task mappings
• SortedList self.st: O(n) to store all task priorities and IDs
• Total auxiliary space: O(n)

Learn more about how to find time and space complexity quickly.

Common Pitfalls

1. Forgetting to Handle the Tie-Breaking Rule Correctly

Pitfall: When multiple tasks have the same priority, the problem requires choosing the one with the highest taskId. A common mistake is to use (priority, taskId) directly in the sorted list, which would give us the lowest taskId when priorities are equal.

Incorrect Implementation:

# Wrong: This would select lowest taskId when priorities are equal
self.priority_queue.add((-priority, taskId))  # INCORRECT!

Solution: Store both priority and taskId as negative values: (-priority, -taskId). This ensures that when the sorted list orders tuples, higher taskIds come first among tasks with the same priority.

2. Using remove() Instead of discard() in Edit Operation

Pitfall: In the edit() method, if there's any synchronization issue or the task was already removed, using remove() will raise a ValueError if the element is not found.

Problematic Code:

def edit(self, taskId: int, newPriority: int) -> None:
    userId, priority = self.task_info[taskId]
    self.priority_queue.remove((-priority, -taskId))  # May raise ValueError!
    # ...

Solution: Use discard() which silently does nothing if the element is not present:

self.priority_queue.discard((-priority, -taskId))  # Safe operation

3. Inconsistent Data Structure Updates

Pitfall: Forgetting to update both data structures (dictionary and sorted list) in sync. For example, removing from one but not the other.

Incorrect Implementation:

def rmv(self, taskId: int) -> None:
    # Forgot to get priority before removing from dictionary!
    self.task_info.pop(taskId)
    # Now we can't remove from sorted list because we lost the priority
    self.priority_queue.remove((-priority, -taskId))  # priority is undefined!

Solution: Always retrieve necessary information before modifying any data structure:

def rmv(self, taskId: int) -> None:
    _, priority = self.task_info[taskId]  # Get priority first
    self.task_info.pop(taskId)
    self.priority_queue.remove((-priority, -taskId))

4. Incorrect Extraction of TaskId in execTop()

Pitfall: Forgetting to negate the taskId when extracting it from the sorted list tuple.

Incorrect Implementation:

def execTop(self) -> int:
    if not self.priority_queue:
        return -1
    task_id = self.priority_queue.pop(0)[1]  # Wrong! This is negative
    # task_id is actually -taskId, will cause KeyError in dictionary lookup
    userId, _ = self.task_info[task_id]  # KeyError!

Solution: Remember to negate the taskId to get the original value:

task_id = -self.priority_queue.pop(0)[1]  # Correct: negate to get positive taskId

5. Not Checking for Empty Queue in execTop()

Pitfall: Attempting to pop from an empty sorted list will raise an IndexError.

Incorrect Implementation:

def execTop(self) -> int:
    # Dangerous: No check for empty queue
    task_id = -self.priority_queue.pop(0)[1]  # IndexError if empty!
    # ...

Solution: Always check if the queue is empty before attempting to pop:

def execTop(self) -> int:
    if not self.priority_queue:  # Check for empty queue
        return -1
    task_id = -self.priority_queue.pop(0)[1]
    # ...