কনটেন্টে যান

Heap — প্রিন্টিং ওয়ার্কশপের কাজের সারি

একটি printer একসঙ্গে একটি কাজ করতে পারে। কাজ আসার সময় ও ছাপতে কত সময় লাগবে জানা আছে। Printer ফাঁকা হলে তখন পর্যন্ত এসে পৌঁছানো কাজগুলোর মধ্যে সবচেয়ে কম সময়ের কাজটি নেবে। পাশাপাশি প্রতিটি শুরু হওয়া কাজের অপেক্ষার সময়ের median দেখাতে হবে।

Microsoft কোর্সের সংযোগ: Module 5-এর priority queue ও optimization-এর অনুশীলন; Module 4-এর greedy সিদ্ধান্ত বোঝাতেও কাজে লাগে। এখানে নির্ধারিত dispatch rule বাস্তবায়ন করবেন; সব পরিস্থিতিতে globally optimal schedule পাওয়ার দাবি নেই।

১. আগে পড়া ও ছোট অনুশীলন

ক্রম উপকরণ প্রস্তুতির প্রশ্ন
১ Heap ধারণা, চিত্র Heap-এর root কী গ্যারান্টি দেয়? পুরো array কি sorted?
২ Heap patterns Tuple priority ও tie-break কীভাবে লিখবেন?
৩ Single-threaded CPU Arrival order ও execution order আলাদা কেন?
৪ Median from data stream দুই heap-এর size ও boundary invariant কী?

প্রথমে ৩ নম্বরের দুই-তিনটি job হাতে চালান। তারপর ৪ নম্বরে চারটি value যোগ করে প্রতিবার median লিখুন। এই দুই ফল একত্র করেই lab-এর report হবে।

২. সমস্যা, input ও output

schedule_report(jobs) লিখুন; প্রতিটি job (id, arrival, duration)।

  1. সময় 0 থেকে printer চালু। চলমান job শেষ না হওয়া পর্যন্ত থামানো যাবে না: এটি non-preemptive scheduling।
  2. Printer ফাঁকা হলে arrival ≤ now job-গুলোর সর্বনিম্ন duration নিন। Duration সমান হলে string ID-এর lexical order নিন।
  3. কোনো job ready না থাকলে পরের arrival-এ clock এগিয়ে নিন।
  4. প্রতিটি job-এর wait = start - arrival। সেই wait-কে data stream-এ যোগ করে median লিখুন। জোড় সংখ্যক value হলে মাঝের দুইটির গড় median।
  5. Schedule, total wait, প্রতি ধাপের median, final median ও শেষ completion time দিন।

সীমা: 0 ≤ n ≤ 100,000, ID আলাদা nonempty string, arrival ও duration পূর্ণসংখ্যা, 0 ≤ arrival ≤ 10⁹, 1 ≤ duration ≤ 10⁶। Input যেকোনো arrival order-এ থাকতে পারে। Duplicate ID, negative arrival বা nonpositive duration-এ ValueError। ফাঁকা input-এ median None, finish 0।

Python
schedule_report([("A", 0, 5), ("B", 1, 2), ("C", 1, 1), ("D", 9, 2)])
Dispatch Start → finish Wait এই পর্যন্ত অপেক্ষার median
A 0 → 5 0 0
C 5 → 6 4 2
B 6 → 8 5 4
D 9 → 11 0 2
Python
{
    "schedule": [
        {"id": "A", "start": 0, "finish": 5, "wait": 0},
        {"id": "C", "start": 5, "finish": 6, "wait": 4},
        {"id": "B", "start": 6, "finish": 8, "wait": 5},
        {"id": "D", "start": 9, "finish": 11, "wait": 0},
    ],
    "total_wait": 9,
    "median_trace": [0, 2, 4, 2],
    "median_wait": 2,
    "finish": 11,
}

Median 2 ও 2.0 একই numeric উত্তর। t=1-এ C এলেও A চলতে থাকবে। t=8 থেকে 9 printer idle থাকবে।

flowchart LR
    arrivals["Arrival অনুযায়ী job"] --> ready["Ready min-heap\nduration, ID"]
    ready --> run["পরের job চালান"]
    run --> wait["start − arrival"]
    wait --> lower["ছোট অর্ধেক\nmax-heap"]
    wait --> upper["বড় অর্ধেক\nmin-heap"]
    lower --> median["দুই root থেকে median"]
    upper --> median

৩. Guided milestones

ধাপ A — Arrival ও clock আলাদা রাখুন

Arrival অনুযায়ী job-এর একটি copy sort করুন; একটি pointer বলবে কোন job এখনও ready heap-এ যায়নি। Ready heap খালি হলেই clock পরের arrival পর্যন্ত এগোতে পারে।

Hint: Clock এক unit করে বাড়াবেন না; arrival 10⁹ হলেও একটি jump যথেষ্ট। Checkpoint: শুধু ("A", 10, 2) হলে start 10, wait 0, finish 12।

ধাপ B — বাস্তব priority queue দিয়ে dispatch

(duration, id, arrival) tuple heap-এ দিন। প্রত্যেক dispatch-এর আগে arrival ≤ now সব job ঢোকান। একটি pop করুন এবং পুরো duration চালান।

Invariant: heap-এ ঠিক সেই job-গুলো আছে যেগুলো এসে গেছে, এখনও চালানো হয়নি। ভবিষ্যতের ছোট কাজ current job-কে থামায় না।

Checkpoint: example-এ t=5-এ heap-এর candidate B ও C; C আগে। শুধু arrival-sorted list চালালে এই ফল আসবে না।

ধাপ C — Wait stream-এর median

একটি max-heap-এ ছোট অর্ধেক, একটি min-heap-এ বড় অর্ধেক রাখুন। Python heapq-তে max-heap-এর জন্য negative value ব্যবহার করতে পারেন। নতুন wait ঢুকিয়ে প্রয়োজনে এক heap-এর root অন্যটিতে সরান।

Invariant: ছোট অর্ধেকের সব value বড় অর্ধেকের value-এর সমান বা ছোট; ছোট heap-এর size বড় heap-এর সমান অথবা এক বেশি।

Checkpoint: wait [0, 4, 5]-এ median 4; পরের 0 এলে sorted wait [0, 0, 4, 5], median 2। প্রতি ধাপে পুরো list sort করে median নেবেন না।

ধাপ D — Report ও audit

Dispatch record থেকে total wait যোগ করুন; ধাপ C-এর median একই record-এর wait থেকে তৈরি হবে। Finish আগের completion থেকে পরের job-এর start নির্ধারণ করবে।

Checkpoint: সব job একবার আছে; প্রতিটি finish - start == duration; printer-এর কাজ overlap করে না; report-এর total_wait == 9।

৪. সমাধানের ব্যাখ্যা

কেন এই data structure-গুলো যথেষ্ট

Arrival list সময়ের প্রবাহ সামলায়। Ready heap প্রতি dispatch-এ বর্তমান candidate-গুলোর সঠিক minimum দেয়। Clock কখনও পিছিয়ে যায় না, তাই একটি job সর্বোচ্চ একবার ready heap-এ ঢোকে ও বের হয়।

Median-এর দুই heap value-গুলোকে দুই ভাগে ভাগ করে। Boundary root দুটো জানা থাকলেই মাঝের value পাওয়া যায়। Size এক ধাপের বেশি অসমান হলে একটি root সরিয়ে balance ফেরানো যায়।

সম্পূর্ণ Python solution · unit test।

৫. পরীক্ষা

Repository root থেকে, অথবা download করা দুই ফাইল practice/ folder-এ রেখে:

Bash
python -m unittest discover -s practice -p 'test_heap_lab.py' -v

৮টি test method worked example, empty input, tie, non-preemption, idle gap, invalid input ও median invariant পরীক্ষা করে। ১০০টি seeded random schedule আলাদা scan-based scheduler এবং statistics.median-এর সঙ্গে মেলানো হয়।

Complexity: arrival sort O(n log n); প্রতিটি job তিনটি heap-এর সীমিত push/pop করে, মোট O(n log n)। Heap ও report মিলিয়ে space O(n)। Median পড়া O(1), নতুন wait যোগ O(log n)।

৬. Coverage ও পরের অনুশীলন

ব্যবহার হয়েছে পরের ধাপে আলাদা করে করবেন
Ready queue, tuple priority, tie-breaking Task scheduler cooldown
দুই heap দিয়ে streaming median Sliding-window median: পুরোনো value মুছতে হয়
Greedy dispatch simulation IPO: eligibility ও profit আলাদা
Heap থেকে minimum বারবার নেওয়া Merge k sorted lists

এই lab scheduling ও median একত্র করে; সব heap problem-এর প্রতিটি pattern এতে নেই।