Stream موازی

بهینه‌سازی محاسبات Stream

یکی از ویژگی‌های هیجان‌انگیز Stream API این است که یک استریم قادر به پردازش موازی داده‌هاست. پردازش موازی داده با Stream API به سادگی فراخوانی متد parallel() روی هر استریم موجود است.

اجرای این کد نتیجه زیر را به شما می‌دهد.

Sum = 45

این مجموع واقعاً به صورت موازی محاسبه شده. البته روی چنین مثال کوچکی ممکن است هیچ بهبود عملکردی متوجه نشوید.

چرا می‌خواهید داده‌ها را به صورت موازی محاسبه کنید؟ احتمالاً برای دریافت سریع‌تر نتیجه‌ی محاسبات. آیا استریم موازی نتیجه‌ی سریع‌تری نسبت به استریم ترتیبی به شما می‌دهد؟ خب، پاسخ به این سؤال به این سادگی نیست. در برخی موارد بله، اما متأسفانه در برخی موارد نه. هر چقدر هم ناامیدکننده به نظر برسد، استریم موازی همیشه سریع‌تر از استریم ترتیبی نیست.

با در نظر گرفتن این موضوع، باید محتاط باشید: انتخاب استفاده از استریم‌های موازی تصمیمی نیست که باید سرسری گرفته شود. چندین سؤال وجود دارد که حتی قبل از در نظر گرفتن موازی‌سازی باید از خود بپرسید.

اول از خود بپرسید، آیا به آن نیاز دارید؟ آیا الزامات عملکردی در برنامه‌ی شما وجود دارد که برآورده نمی‌شوند؟ آیا مطمئنید مشکلات عملکردی شما از پردازش استریمی نشأت می‌گیرند که می‌خواهید محاسبه کنید؟ چگونه قصد دارید بهبود عملکرد خود را اندازه‌گیری کنید تا مطمئن شوید رفتن به حالت موازی برای این محاسبه‌ی خاص عملکرد برنامه شما را بهبود بخشیده؟

رفتن به حالت موازی توان محاسباتی بیشتری مصرف می‌کند. آیا CPU یا هسته‌ی CPU اضافی دارید که بتوانید به این محاسبه بدهید؟ آیا می‌توانید چرخه‌های CPU بیشتری به محاسبه بدهید بدون اینکه بقیه‌ی برنامه کند شود؟

رفتن به حالت موازی thread مصرف می‌کند. آیا thread اضافی دارید که بتوانید به محاسبه بدهید؟ اگر روی برنامه‌ای کار می‌کنید که در یک وب‌سرور اجرا می‌شود، thread های شما برای سرویس‌دهی به درخواست‌های HTTP استفاده می‌شوند. آیا با گرفتن آن‌ها برای کارهای دیگر مشکلی ندارید؟

وقتی تصمیم به موازی‌سازی گرفتید، باید مطمئن شوید عملکرد محاسبات استریم شما واقعاً بهبود یافته. باید این بهبود عملکرد را در بافتی که تا حد امکان به محیط production شما نزدیک است اندازه‌گیری کنید.

در این آموزش، چندین عنصر کلیدی را پوشش می‌دهیم که به شما کمک می‌کند بهبودهای مورد انتظار از موازی‌سازی را ارزیابی کنید و برخی عناصر دیگر که باید شما را در مورد موازی‌سازی محتاط کنند. اما در نهایت، تنها چیزی که باید به شما بگوید موازی‌سازی ارزشش را دارد یا نه، تست و اندازه‌گیری زمان اجرا است.

پیاده‌سازی موازی‌سازی

موازی‌سازی در Stream API با استفاده از تجزیه‌ی بازگشتی (recursive decomposition) داده‌هایی که استریم پردازش می‌کند پیاده‌سازی شده. روی Fork/Join Framework که در JDK 7 اضافه شده ساخته شده.

تجزیه شامل تقسیم داده‌هایی که استریم شما پردازش می‌کند به دو بخش است. هر بخش سپس توسط هسته‌ی CPU خودش پردازش می‌شود که ممکن است تصمیم بگیرد آن را دوباره به صورت بازگشتی تقسیم کند.

در نقطه‌ای، فریمورک تصمیم می‌گیرد مقدار داده در یک بخش مشخص به اندازه‌ی کافی کوچک است که به صورت عادی پردازش شود. این زیرمجموعه داده سپس پردازش و یک نتیجه‌ی جزئی محاسبه می‌شود. این نتیجه‌ی جزئی سپس با نتایج جزئی دیگری که از بخش‌های دیگر روی هسته‌های CPU دیگر محاسبه شده ادغام می‌شود.

رفتن به حالت موازی overhead دارد. این overhead باید در مقایسه با مزایای توزیع محاسبات روی چندین هسته CPU کوچک باشد. در غیر این صورت، موازی‌سازی عملکرد محاسبات شما را بدتر می‌کند به جای بهتر کردن آن.

بیایید تمام این مراحل را یکی‌یکی بررسی کنیم و ببینیم چه چیزی می‌تواند مانع بهبود عملکرد شما شود.

درک مکان‌گرایی داده (Data Locality)

مکان‌گرایی داده بر سرعت پردازش داده‌های شما تأثیر دارد، چه به صورت ترتیبی و چه موازی پردازش شود. هرچه مکان‌گرایی بهتر باشد، محاسبات سریع‌تر خواهد بود.

برای در دسترس قرار گرفتن CPU، داده‌های شما باید از حافظه اصلی کامپیوتر به کش CPU منتقل شوند. از نظر فیزیکی، حافظه اصلی یک جزء مجزا از CPU شماست. از طرف دیگر، کش همان سیلیکون CPU را با اجزای محاسباتی اصلی به اشتراک می‌گذارد. انتقال داده از حافظه اصلی به کش CPU در مقایسه با سرعتی که هسته‌ی CPU می‌تواند از کش خود دسترسی داشته باشد بسیار کند است.

وقتی CPU شما به داده نیاز دارد، ابتدا بررسی می‌کند آیا این داده در کش موجود است. اگر باشد، می‌تواند فوراً از آن استفاده کند. اگر نباشد، این داده باید از حافظه اصلی واکشی و در کش کپی شود. این وضعیت cache miss نامیده می‌شود. Cache miss ها پرهزینه هستند چون در این مدت CPU شما منتظر داده است. می‌خواهید از این وضعیت اجتناب کنید.

نحوه‌ی انتقال داده بین حافظه اصلی و کش CPU نقش مهمی در جلوگیری از cache miss دارد. حافظه به صورت خطوط (lines) سازمان‌دهی شده. معمولاً یک خط ۶۴ بایت طول دارد، یعنی هشت مقدار long. تمام انتقالات بین حافظه اصلی و کش CPU به صورت خط‌به‌خط انجام می‌شود. پس حتی اگر CPU فقط به یک مقدار int نیاز داشته باشد، کل خطی که آن مقدار را شامل می‌شود به کش انتقال می‌یابد.

تکرار روی آرایه‌ی انواع اولیه

فرض کنید کد شما روی آرایه‌ی نوع int[] تکرار می‌کند. یک خط ۶۴ بایتی می‌تواند ۱۶ مقدار int را در خود جای دهد. فرض کنید دسترسی به اولین عنصر آرایه‌ی شما یک cache miss باشد. CPU سپس خطی که شامل این عنصر است را در کش خود بارگذاری می‌کند تا تکرار شروع شود. چون یک خط کامل بارگذاری کرده، ۱۵ مقدار بعدی نیز ممکن است انتقال یافته باشند. دسترسی به مقادیر بعدی بسیار سریع خواهد بود.

در آن صورت، mکان‌گرایی داده عالی است: داده‌های شما فیزیکاً در نواحی مجاور حافظه اصلی ذخیره شده‌اند. این مطلوب است چون انتقال داده از حافظه اصلی به کش CPU بسیار سریع‌تر خواهد بود.

تکرار روی آرایه‌ی نمونه‌های Integer

اکنون فرض کنید کد شما روی آرایه‌ی نوع Integer[] تکرار می‌کند. آنچه واقعاً دارید دیگر آرایه‌ای از انواع اولیه نیست بلکه آرایه‌ای از ارجاعات است. هر سلون این آرایه حاوی ارجاعی به شیء Integer است که می‌تواند هر جای حافظه باشد.

اگر دسترسی به اولین عنصر آرایه‌ی شما یک cache miss باشد، CPU باید خطی که شامل این عنصر است را به کش خود بارگذاری کند. آنچه واقعاً بارگذاری می‌کند ۱۶ ارجاع اول آرایه‌ی شماست. سپس باید اولین شیء Integer را بارگذاری کند که ممکن است جای دیگری در حافظه اصلی باشد و به cache miss دیگری منجر شود. در واقع، احتمالاً هر خواندن هر شیء Integer از آرایه‌ی شما نیز cache miss خواهد بود.

در آن صورت، mکان‌گرایی داده به خوبی مثال قبلی نیست: ارجاعات به داده‌های شما فیزیکاً در نواحی مجاور حافظه اصلی ذخیره شده‌اند، اما مقادیری که برای انجام محاسبات نیاز دارید نه.

تکرار روی لیست پیوندی نمونه‌های Integer

بیایید یک وضعیت آخر را بررسی کنیم. فرض کنید کد شما روی لیستی از نوع LinkedList<Integer> تکرار می‌کند. اگر دسترسی به اولین عنصر cache miss باشد، CPU اولین گره لیست پیوندی شما را به کش خود بارگذاری می‌کند. آن گره دو ارجاع شامل می‌شود: اولی به مقداری که برای محاسبات نیاز دارید، دومی به گره بعدی لیست. این وضعیت از قبلی بدتر است: احتمالاً دسترسی به مقدار بعدی لیست شما دو cache miss ایجاد می‌کند.

در این مورد، mکان‌گرایی داده وحشتناک است: نه داده‌های شما و نه ارجاعات به آن‌ها در نواحی مجاور حافظه اصلی ذخیره شده‌اند.

اجتناب از Pointer Chasing

نیاز به دنبال کردن ارجاعات یا اشاره‌گرها برای رسیدن به عنصر صحیحی که داده‌ی مورد نیاز شما را حمل می‌کند pointer chasing نامیده می‌شود. Pointer chasing چیزی است که می‌خواهید در برنامه‌ی خود از آن اجتناب کنید و منبع بسیاری از hits عملکردی است. Pointer chasing در تکرار روی آرایه‌ای از مقادیر int وجود ندارد. این hit اصلی عملکردی شما هنگام تکرار روی لیست پیوندی نمونه‌های Integer است.

تجزیه‌ی یک منبع داده

اگر تصمیم به پردازش موازی یک استریم دارید، اولین مرحله شامل تجزیه‌ی منبع داده شماست. برای اینکه تجزیه کارآمد باشد، باید چندین ویژگی داشته باشد.

• تجزیه‌ی ساختار داده باید آسان و سریع باشد.
• تجزیه باید یکنواخت باشد: دو substream که دریافت می‌کنید باید مقدار یکسانی داده برای پردازش داشته باشند.

[دیاگرام: مقایسه‌ی تجزیه‌پذیری ArrayList، LinkedList، HashSet و TreeSet]

هر ساختاری که در Collection Framework استفاده می‌شود رابط Collection را پیاده‌سازی می‌کند. پس پیاده‌سازی استریم می‌تواند با فراخوانی ساده‌ی size() تعداد عناصری را که باید پردازش کند بدست آورد.

دسترسی به تعداد عناصری که یک استریم باید پردازش کند همیشه ممکن نیست. مثلاً می‌توانید استریمی روی خطوط یک فایل متنی ایجاد کنید. پیاده‌سازی استریم نمی‌داند چند خط باید پردازش کند.

تجزیه‌ی خطوط یک فایل متنی

الگوی Files.lines(path) که قبلاً پوشش دادیم استریمی ایجاد می‌کند که خطوط فایل متنی را پردازش می‌کند. بدون تحلیل فایل، دریافت تعداد خطوط فایل متنی ممکن نیست.

همین مورد برای الگوی Pattern.splitAsStream(line) نیز صادق است.

تجزیه‌ی بازه یا استریم تولیدی

استریم‌های تخصصی اعداد الگوهایی برای ایجاد استریم‌ها به شما می‌دهند.

IntStream.range(0, 10) به سادگی تجزیه‌پذیر است. اما متدهای Stream.generate() و Stream.iterate() منبع داده‌ی به راحتی تجزیه‌پذیری به شما نمی‌دهند. در واقع، این منبع ممکن است نامتناهی باشد و فقط توسط نحوه‌ی پردازش آن در استریم محدود شود.

تجزیه و توزیع کار

وقتی منبع داده‌ی شما تجزیه شد، دو substream باید روی هسته‌های مختلف CPU شما پردازش شوند تا موازی‌سازی مؤثر باشد.

این کار توسط Fork/Join Framework انجام می‌شود. Fork/Join Framework یک pool از thread ها را مدیریت می‌کند، که هنگام اجرای برنامه ایجاد می‌شوند و Common Fork/Join Pool نامیده می‌شوند. تعداد thread ها در این pool با تعداد هسته‌های CPU شما هماهنگ است.

1. اولین thread pool اولین task را ایجاد می‌کند. اجرای این task تصمیم می‌گیرد آیا محاسبه به اندازه‌ی کافی کوچک است که به صورت ترتیبی محاسبه شود یا بزرگ است و باید تقسیم شود.

2. اگر تقسیم شود، دو subtask ایجاد و در صف آن thread ذخیره می‌شوند. task اصلی سپس منتظر تکمیل دو subtask می‌ماند. در حین انتظار، خودش نیز در این صف انتظار ذخیره می‌شود.

3. اگر محاسبه انجام شود، یک نتیجه تولید می‌شود. این نتیجه یک نتیجه‌ی جزئی از کل محاسبه است. این task سپس نتیجه را به task اصلی که آن را ایجاد کرده برمی‌گرداند.

4. وقتی یک task دو نتیجه از دو subtask ایجادشده داشت، می‌تواند آن‌ها را ادغام کند تا نتیجه تولید و به task اصلی برگرداند.

در نقطه‌ی مشخصی، اولین task اصلی دو نتیجه‌ی جزئی از دو subtask خود دریافت می‌کند. سپس قادر به ادغام آن‌ها و برگرداندن نتیجه‌ی نهایی محاسبه است.

فعلاً تنها thread‌ای که کار می‌کند اولین thread pool است که توسط Fork/Join Framework فراخوانی شده. Fork/Join Framework الگوی دیگری از برنامه‌نویسی همزمان به نام work stealing (ربودن کار) پیاده‌سازی می‌کند. یک thread بیکار pool می‌تواند صف انتظار thread های دیگر همان pool را بررسی کند و task‌ای ببرد و پردازش کند.

این اتفاقی است که در این مورد رخ می‌دهد. به محض اینکه تعداد task ها در صف انتظار اول زیاد شود، thread های دیگر some از آن‌ها را می‌ربایند، پردازش می‌کنند، کار را بیشتر تقسیم و صف‌های انتظار خودشان را با task های بیشتر پر می‌کنند. این ویژگی تمام thread های pool را مشغول نگه می‌دارد.

این ویژگی work stealing خوب کار می‌کند اما یک نقطه ضعف دارد: بسته به نحوه‌ی تجزیه‌ی منبع و نحوه‌ی جابه‌جایی task ها از یک thread به دیگری، داده‌های شما ممکن است به هر ترتیبی پردازش شوند. این در برخی موارد می‌تواند مشکل‌ساز باشد.

پردازش یک Substream

پردازش یک substream می‌تواند با پردازش یک استریم کامل متفاوت باشد. دو عنصر می‌توانند پردازش یک substream را متفاوت کنند: دسترسی به یک state خارجی، و حمل state از پردازش یک عنصر به عنصر دیگر. این دو عنصر بر عملکرد استریم‌های موازی شما تأثیر خواهند گذاشت.

دسترسی به state خارجی

Fork/Join Framework محاسبه‌ی شما را به subtask های متعددی تقسیم می‌کند، هر کدام توسط thread‌ای از pool پردازش می‌شود.

اگر استریم را به صورت ترتیبی پردازش کنید، تمام عناصر در thread‌ای پردازش می‌شوند که متد شما را اجرا می‌کند. اگر همان استریم را به صورت موازی پردازش کنید، عناصر توسط thread‌ای از Common Fork/Join pool پردازش می‌شوند.

دسترسی به state خارج از استریم شما سپس از thread دیگری انجام می‌شود و می‌تواند به race condition منجر شود.

بیایید کد زیر را در نظر بگیرید. متأسفانه نمی‌توانید این کد را در مرورگر خود اجرا کنید، باید آن را کپی و در IDE خود جایگذاری کنید تا ببینید چگونه کار می‌کند.

نتیجه‌ای که تولید می‌کند به صورت زیر است.

Thread names:
main

اگر فراخوانی parallel() را فعال کنید، این استریم به صورت موازی اجرا می‌شود. نتیجه به صورت زیر تغییر می‌کند و ممکن است در ماشین شما متفاوت باشد.

Thread names:
ForkJoinPool.commonPool-worker-3
ForkJoinPool.commonPool-worker-4
ForkJoinPool.commonPool-worker-2
ForkJoinPool.commonPool-worker-4
main
ForkJoinPool.commonPool-worker-5

هر دسترسی به عنصر خارجی غیرهمزمان می‌تواند به race condition و عدم ثبات داده منجر شود. بیایید کد زیر را اجرا کنیم.

اجرای چند باره‌ی این کد می‌تواند به نتایج مختلفی منجر شود چون تمام thread های Common Fork/Join Pool سعی می‌کنند همزمان داده به نمونه‌ای از ArrayList اضافه کنند، که ساختار thread-safe نیست. احتمال دیدن نتیجه‌ی صحیح کم است و حتی ممکن است ArrayIndexOutOfBoundsException دریافت کنید. اجرای این نوع کد با هر collection یا map غیرهمزمانی به نتایج غیرقابل پیش‌بینی منجر می‌شود، از جمله exception ها.

یک اجرای معمولی چیزی شبیه به این به شما می‌دهد. بله، عناصر از دست‌رفته‌ی زیادی وجود دارد!

ints.size() = 387122

تغییر state خارج از استریم از داخل آن یک antipattern است.

ترتیب مواجهه (Encounter Order)

مواردی وجود دارد که ترتیب پردازش داده در Stream API مهم است. این مورد برای متدهای زیر صادق است.

• limit(n): پردازش را به n اولین عنصر این استریم محدود می‌کند.
• skip(n): پردازش n اولین عنصر این استریم را رد می‌کند.
• findFirst(): اولین عنصر استریم را پیدا می‌کند.

این سه متد باید ترتیب پردازش عناصر استریم را به یاد بیاورند و عناصر را بشمارند تا نتیجه‌ی صحیح تولید کنند.

آن‌ها stateful (با state) نامیده می‌شوند، چون برای کار کردن به state داخلی نیاز دارند.

داشتن چنین عملیات‌های stateful منجر به overhead در استریم‌های موازی می‌شود. مثلاً limit() برای کار صحیح به یک شمارنده‌ی داخلی نیاز دارد. در حالت موازی، این شمارنده‌ی داخلی بین thread های مختلف مشترک است. اشتراک‌گذاری state قابل تغییر بین thread ها پرهزینه است و باید از آن اجتناب شود.

درک overhead پردازش موازی یک Stream

پردازش موازی یک استریم محاسباتی برای مدیریت موازی‌سازی اضافه می‌کند. این عناصر هزینه دارند و باید آن‌ها را بشناسید تا مطمئن شوید این هزینه بیش از حد بالا نخواهد بود در مقایسه با مزایای رفتن به حالت موازی.

• داده‌های شما باید تجزیه شوند. تجزیه می‌تواند ارزان باشد و می‌تواند پرهزینه باشد، بسته به داده‌هایی که پردازش می‌کنید. مکان‌گرایی بد برای داده‌های شما تجزیه را پرهزینه می‌کند.
• تجزیه باید کارآمد باشد. باید substream های یکنواخت ایجاد کند. برخی منابع به راحتی یکنواخت تجزیه می‌شوند، برخی نه.
• پس از تجزیه، پیاده‌سازی داده‌های شما را همزمان پردازش می‌کند. باید از هر دسترسی به هر state خارجی قابل تغییر اجتناب کنید و همچنین از داشتن state داخلی مشترک قابل تغییر.
• سپس نتایج جزئی باید ادغام شوند. نتایجی هستند که به راحتی قابل ادغام هستند. ادغام مجموع اعداد صحیح آسان و ارزان است. ادغام collection ها نیز آسان است. ادغام hashmap ها پرهزینه‌تر است.

بیان قوانینی برای استفاده‌ی صحیح از استریم‌های موازی

قانون ۱: بهینه‌سازی نکنید چون سرگرم‌کننده است؛ بهینه‌سازی کنید چون الزاماتی دارید که برآورده نمی‌شوند.

قانون ۲: منبع داده‌ی خود را با احتیاط انتخاب کنید.

قانون ۳: state خارجی را تغییر ندهید و state قابل تغییر مشترک نداشته باشید.

قانون ۴: حدس نزنید؛ عملکرد کد خود را اندازه‌گیری کنید.

اجرای این کد چندین بار ممکن است نتایج متفاوتی تولید کند چون تمام thread های Common Fork/Join Pool سعی می‌کنند همزمان داده را در نمونه‌ای از ArrayList اضافه کنند که ساختار thread-safe نیست. احتمال دیدن نتیجه‌ی درست کم است و حتی ممکن است ArrayIndexOutOfBoundsException دریافت کنید.

یک اجرای معمولی چیزی شبیه این به شما می‌دهد. بله، عناصر گمشده‌ی زیادی وجود دارد!

ints.size() = 387122

تغییر state خارج از استریم از داخل آن یک antipattern است.