This is the roughly 40-minute coding portion of a Palantir phone screen. Unlike Palantir's onsite learning and code review rounds, this is a self-contained problem: you are given a template class and asked to fill in its methods. You are also expected to write your own test cases once the implementation is in place.
Implement a session manager class that assigns incoming sessions to a fixed set of servers while keeping the load balanced across those servers. The class exposes two methods:
def start_session(self, session_id: str) -> None:
...
def get_allocation(self) -> Dict[str, Set[str]]:
...
start_session places a new session on one of the servers. get_allocation returns the current mapping from each server id to the set of session ids currently assigned to it. [Source: darkinterview.com]
The manager owns a fixed list of servers, for example [s1, s2, s3]. As sessions are added, the number of sessions on any server must never differ from another server's count by more than 1. With 24 sessions across three servers the allocation should look like [8, 8, 8]; with 23 it should look like [8, 8, 7]. An allocation such as [9, 8, 6] is invalid because the gap between the busiest and least busy server exceeds 1.
The way to satisfy this is to always assign the next session to a least-loaded server. As long as each new session lands on a server that currently holds the minimum count, the spread between any two servers stays within 1.
A min heap of (num_of_sessions, server_id) makes each assignment cheap. On every start_session: [Source: darkinterview.com]
Because the heap always surfaces the server with the fewest sessions, assignments stay balanced without scanning every server each time.
import heapq
from typing import Dict, Set
class SessionManager:
def __init__(self, servers):
self._heap = [(0, server_id) for server_id in servers]
heapq.heapify(self._heap)
self._allocation: Dict[str, Set[str]] = {s: set() for s in servers}
self._seen: Set[str] = set()
def start_session(self, session_id: str) -> None:
if session_id in self._seen:
return
count, server_id = heapq.heappop(self._heap)
self._allocation[server_id].add(session_id)
self._seen.add(session_id)
heapq.heappush(self._heap, (count + 1, server_id))
def get_allocation(self) -> Dict[str, Set[str]]:
return self._allocation
A session id that has already been started must not be added again. Without a guard, calling start_session twice with the same id would consume two slots for what is really one session and could push the allocation out of balance. Track the set of session ids that have already been seen, and make start_session a no-op when the id is already present.
You are expected to write the tests yourself, covering both normal operation and the edge case: [Source: darkinterview.com]
[8, 8, 7] in some order.