রুবি ব্যাকগ্রাউন্ড প্রসেসিং সিস্টেম বানান: স্ক্র্যাচ থেকে শিখুন

আজকের পোস্টে, আমরা একটি নাইভ ব্যাকগ্রাউন্ড প্রসেসিং সিস্টেম ইমপ্লিমেন্ট করব ফান করার জন্য! আমরা পথে কিছু শিখতে পারি পপুলার ব্যাকগ্রাউন্ড প্রসেসিং সিস্টেম যেমন Sidekiq-এর ইন্টারনালে ঝাঁকি দিয়ে। এই ফানের ফলাফল بأي কোনো প্রোডাকশন ইউজের জন্য নন।

চলুন কল্পনা করি আমাদের অ্যাপ্লিকেশনে একটি টাস্ক আছে যা এক বা একাধিক ওয়েবসাইট লোড করে এবং তাদের টাইটেল এক্সট্রাক্ট করে। যেহেতু আমাদের এই ওয়েবসাইটগুলোর পারফরম্যান্সের ওপর কোনো নিয়ন্ত্রণ নেই, আমরা টাস্কটি আমাদের মেইন থ্রেডের বাইরে (বা করেন্ট রিকোয়েস্টের বাইরে — যদি আমরা ওয়েব অ্যাপ্লিকেশন বানাই), কিন্তু ব্যাকগ্রাউন্ডে করা চাই।

একটি টাস্ক এনক্যাপসুলেট করা

ব্যাকগ্রাউন্ড প্রসেসিং-এ যাওয়ার আগে, আসুন একটি সার্ভিস অবজেক্ট বানাই টাস্কটি পারফর্ম করার জন্য। আমরা OpenURI এবং Nokogiri ব্যবহার করব টাইটেল ট্যাগের কন্টেন্ট এক্সট্রাক্ট করার জন্য।

require 'open-uri' require 'nokogiri' class TitleExtractorService def call(url) document = Nokogiri::HTML(open(url)) title = document.css('html > head > title').first.content puts title.gsub(/[[:space:]]+/, ' ').strip rescue puts "Unable to find a title for #{url}" end end

সার্ভিসটি কল করলে দেওয়া URL-এর টাইটেল প্রিন্ট হয়।

TitleExtractorService.new.call('https://appsignal.com') # AppSignal: Application Performance Monitoring for Ruby on Rails and Elixir

এটি এক্সপেক্টেড्ली কাজ করে, কিন্তু আসুন দেখি সিনট্যাক্স কিছুটা উন্নত করা যায় কিনা যাতে অন্যান্য ব্যাকগ্রাউন্ড প্রসেসিং সিস্টেমের মত লাগে এবং মনে হয়। একটি Magique::Worker মডিউল তৈরি করে, আমরা সার্ভিস অবজেক্টে কিছু সিনট্যাক্টিক চিনি যোগ করতে পারি।

module Magique module Worker def self.included(base) base.extend(ClassMethods) end module ClassMethods def perform_now(*args) new.perform(*args) end end def perform(*) raise NotImplementedError end end end

মডিউলটি ওয়ার্কার ইনস্ট্যান্সে perform মেথড এবং ওয়ার্কার ক্লাসে perform_now মেথড যোগ করে ইনভোকেশন কিছুটা बेटার করে।

মডিউলটি আমাদের সার্ভিস অবজেক্টে ইনক্লুড করি। একই সাথে, আসুন এটাকে TitleExtractorWorker-এ রিনেম করি এবং call মেথডকে perform-এ পরিবর্তন করি।

class TitleExtractorWorker include Magique::Worker def perform(url) document = Nokogiri::HTML(open(url)) title = document.css('html > head > title').first.content puts title.gsub(/[[:space:]]+/, ' ').strip rescue puts "Unable to find a title for #{url}" end end

ইনভোকেশন একই ফলাফল দেয়, কিন্তু এটি কিছুটা ক্লিয়ার어가 কী হচ্ছে।

TitleExtractorWorker.perform_now('https://appsignal.com') # AppSignal: Application Performance Monitoring for Ruby on Rails and Elixir

অ্যাসিঙ্ক্রোনাস প্রসেসিং ইমপ্লিমেন্ট করা

এখন যেহেতু টাইটেল এক্সট্রাকশন কাজ করছে, আমরা গত Ruby Magic আর্টিকেলগুলোর সব টাইটেল বের করতে পারি। এজন্য, আসুন ধরি আমাদের একটি RUBYMAGIC কনস্ট্যান্ট আছে যেখানে গত আর্টিকেলগুলোর সব URL আছে।

RUBYMAGIC.each do |url| TitleExtractorWorker.perform_now(url) end # Unraveling Classes, Instances and Metaclasses in Ruby | AppSignal Blog # Bindings and Lexical Scope in Ruby | AppSignal Blog # Building a Ruby C Extension From Scratch | AppSignal Blog # Closures in Ruby: Blocks, Procs and Lambdas | AppSignal Blog # ...

আমরা গত আর্টিকেলগুলোর টাইটেল পেলাম, কিন্তু সব এক্সট্রাক্ট করতে সময় লাগল। কারণ আমরা প্রতিটি রিকোয়েস্ট সম্পূর্ণ হওয়ার জন্য অপেক্ষা করতাম পরবর্তীটিতে যাওয়ার আগে।

এটি উন্নত করি perform_async মেথড আমাদের ওয়ার্কার মডিউলে যোগ করে। গতি বাড়াতে, এটি প্রতিটি URL-এর জন্য একটি নতুন থ্রেড তৈরি করে।

module Magique module Worker module ClassMethods def perform_async(*args) Thread.new { new.perform(*args) } end end end end

ইনভোকেশন TitleExtractorWorker.perform_async(url)-এ পরিবর্তন করার পর, আমরা почти একসাথে সব টাইটেল পেলাম। কিন্তু এর মানে এটাও যে আমরা ২০টির অধিক কানেকশন খুললাম Ruby Magic ব্লগে একসাথে। (আপনার ব্লগ নষ্ট করার জন্য দুঃখিত! 😅)

আপনি যদি নিজের ইমপ্লিমেন্টেশন দিয়ে ফলো করছেন এবং লং-রানিং প্রসেস (যেমন ওয়েব সার্ভার) বাইরে টেস্ট করছেন, তো আপনার স্ক্রিপ্টের শেষে loop { sleep 1 } এর মত কিছু যোগ করা ভুলবেন না যাতে প্রসেস তাৎক্ষণিক টার্মিনেট না হয়।

টাস্ক কিউ করা

প্রতিটি ইনভোকেশনের জন্য নতুন থ্রেড তৈরি করার পদ্ধতিতে, আমরা সম্পদ সীমা হিট করব (আমাদের সাইডে এবং যেসব ওয়েবসাইট অ্যাক্সেস করছি সেগুলোতে)। যেহেতু আমরা ভালো নাগরিক হতে চাই, আসুন ইমপ্লিমেন্টেশন পরিবর্তন করি এমন কিছুতে যা অ্যাসিঙ্ক্রোনাস কিন্তু ডিনায়াল-অফ-সার্ভিস আটাকের মত মনে হয় না।

এই সমস্যা সমাধানের একটি সাধারণ পদ্ধতি হলো প্রডিউসার/কনসিউমার প্যাটার্ন। এক বা একাধিক প্রডিউসার টাস্ক কিউতে পুশ করে যখন এক বা একাধিক কনসিউমার কিউ থেকে টাস্ক নিয়ে প্রসেস করে।

একটি কিউ মূলত এলিমেন্টের একটি লিস্ট।理論上, একটি সিম্পল অ্যারে কাজ করবে। কিন্তু, যেহেতু আমরা কনকারেন্সি নিয়ে কাজ করছি, আমাদের নিশ্চিত করতে হবে যে শুধু এক প্রডিউসার বা কনসিউমার একবারে কিউ অ্যাক্সেস করতে পারে। আমরা সাবধান না হলে, বিষয়টি ক্যउस트에 শেষ হবে — ঠিক যেমন দুজন লোক একসাথে এক দরজা দিয়ে বের হতে চাই।

এই সমস্যাকে প্রডিউসার-কনসিউমার সমস্যা বলা হয় এবং এর একাধিক সমাধান আছে। خوشবুদ্ধি, এটি একটি খুব সাধারণ সমস্যা এবং রুবি একটি প্রপার Queue ইমপ্লিমেন্টেশন দিয়ে আসে যা আমরা থ্রেড সিঙ্ক্রোনাইজেশন নিয়ে চিন্তা না করেই ব্যবহার করতে পারি।

এটি ব্যবহার করতে, আসুন নিশ্চিত করি প্রডিউসার এবং কনসিউমার উভয়ই কিউ অ্যাক্সেস করতে পারে। আমরা আমাদের Magique মডিউলে একটি ক্লাস মেথড যোগ করি এবং একটি Queue ইনস্ট্যান্স অ্যাসাইন করি।

module Magique def self.backend @backend end def self.backend=(backend) @backend = backend end end Magique.backend = Queue.new

পরবর্তী, আমরা আমাদের perform_async ইমপ্লিমেন্টেশন পরিবর্তন করি টাস্ক কিউতে পুশ করার জন্য নовый থ্রেড তৈরি করার বদলে। একটি টাস্ক রিপ্রেজেন্ট করা হয় একটি হ্যাশ হিসেবে যাতে ওয়ার্কার ক্লাসের রেফারেন্স এবং perform_async মেথডে পাস করা আর্গুমেন্ট থাকে।

module Magique module Worker module ClassMethods def perform_async(*args) Magique.backend.push(worker: self, args: args) end end end end

এতে, প্রডিউসার সাইডের কথা শেষ। এখন কনসিউমার সাইডে নজর দেই।

প্রতিটি কনসিউমার একটি আলাদা থ্রেড যা কিউ থেকে টাস্ক নিয়ে সেগুলো পারফর্ম করে। এক টাস্ক করার পর থামার বদলে, কনসিউমার কিউ থেকে আরেকটি টাস্ক নিয়ে সেটা পারফর্ম করে, এবং így সیلসিলে। এখানে একটি বেসিক ইমপ্লিমেন্টেশন Magique::Processor নামে। প্রতিটি প্রসেসর একটি নতুন থ্রেড তৈরি করে যা Infinity লুপ করে। প্রতিটি ইটারেশনে, এটি কিউ থেকে একটি নতুন টাস্ক গ্র্যাব করার চেষ্টা করে, ওয়ার্কার ক্লাসের একটি নতুন ইনস্ট্যান্স তৈরি করে, এবং এর perform মেথড কল করে দেওয়া আর্গুমেন্টসহ।

module Magique class Processor def self.start(concurrency = 1) concurrency.times { |n| new("Processor #{n}") } end def initialize(name) thread = Thread.new do loop do payload = Magique.backend.pop worker_class = payload[:worker] worker_class.new.perform(*payload[:args]) end end thread.name = name end end end

প্রসেসিং লুপের পাশাপাশি, আমরা একটি কনভেনিয়েন্স মেথড যোগ করলাম Magique::Processor.start নামে। এটি আমাদের একসাথে একাধিক প্রসেসর স্পিন আপ করতে দেয়। থ্রেড নাম দেওয়া স্ট্রিক্টলি জরুরি নয়, কিন্তু এতে আমরা দেখতে পাব কীভাবে জিনিসপত্র কাজ করছে।

আমাদের TitleExtractorWorker-এর আউটপুট এডজাস্ট করি করেন্ট থ্রেডের নাম ইনক্লুড করতে।

puts "[#{Thread.current.name}] #{title.gsub(/[[:space:]]+/, ' ').strip}"

আমাদের ব্যাকগ্রাউন্ড প্রসেসিং সেটআপ টেস্ট করতে, আমাদের প্রথমে প্রসেসরগুলো স্পিন আপ করতে হবে টাস্কগুলো ইনকিউ করার আগে।

Magique.backend = Queue.new Magique::Processor.start(5) RUBYMAGIC.each do |url| TitleExtractorWorker.perform_async(url) end # [Processor 3] Bindings and Lexical Scope in Ruby | AppSignal Blog # [Processor 4] Building a Ruby C Extension From Scratch | AppSignal Blog # [Processor 1] Unraveling Classes, Instances and Metaclasses in Ruby | AppSignal Blog # [Processor 0] Ruby's Hidden Gems, StringScanner | AppSignal Blog # [Processor 2] Fibers and Enumerators in Ruby: Turning Blocks Inside Out | AppSignal Blog # [Processor 4] Closures in Ruby: Blocks, Procs and Lambdas | AppSignal Blog # ...

এটি রান করলে, আমরা এখনো সব আর্টিকেলের টাইটেল পাই। যদিও এটি প্রতিটি টাস্কের জন্য আলাদা থ্রেড ব্যবহার করার মতো ফাস্ট নয়, এটি প্রাথমিক ইমপ্লিমেন্টেশন থেকে ফাস্ট যেটিতে কোনো ব্যাকগ্রাউন্ড প্রসেসিং ছিল না। যোগ করা প্রসেসর নামের فضلেই, আমরা কনফার্ম করতে পারি সব প্রসেসর কিউর মাধ্যমে কাজ করছে। কনকারেন্ট প্রসেসরের সংখ্যা টিউন করে, প্রসেসিং স্পিড এবং এক্সিস্টিং রিসোর্স লিমিটেশনের মধ্যে একটি Баланс খুঁজে পেতে পারা যায়।

একাধিক প্রসেস এবং মেশিনে বিস্তার

এ পর্যন্ত, আমাদের ব্যাকগ্রাউন্ড প্রসেসিং সিস্টেমের বর্তমান ইমপ্লিমেন্টেশন काफी ভালো কাজ করে। এটি এখনো একই প্রসেসে সীমাবদ্ধ, যাইহোক। রিসোর্স-হাঙ্গরি টাস্ক পুরো প্রসেসের পারফরম্যান্সকে প্রভাবিত করবে। শেষ ধাপ হিসেবে, আসুন দেখি ওয়ার্কলোড কীভাবে একাধিক প্রসেস এবং হয়তো একাধিক মেশিনে ডিস্ট্রিবিউট করা যায়।

কিউ হলো প্রডিউসার এবং কনসিউমারের একমাত্র সংযোগ। এখনো, এটি ইন-মেমরি ইমপ্লিমেন্টেশন ব্যবহার করছে। Sidekiq থেকে আরও অনুপ্রেরণা নিয়ে আসি এবং Redis ব্যবহার করে একটি কিউ ইমপ্লিমেন্ট করি।

Redis-এর লিস্টের সাপোর্ট আছে যা আমাদের টাস্ক পুশ এবং ফেচ করতে দেয়। তদুপরি, Redis রুবি জেম থ্রেড-সেফ এবং লিস্ট মডিফাই করার Redis কমান্ডগুলো অ্যাটমিক। এই প্রোপার্টিগুলো এটি আমাদের অ্যাসিঙ্ক্রোনাস ব্যাকগ্রাউন্ড প্রসেসিং সিস্টেমে ব্যবহার করার জন্য উপযোগী করে সিঙ্ক্রোনাইজেশন সমস্যা ছাড়া।

আসুন একটি Redis-backed কিউ তৈরি করি যা push এবং shift মেথড ইমপ্লিমেন্ট করে ঠিক Queue-এর মত যেটি আমরা আগে ব্যবহার করেছিলাম।

require 'json' require 'redis' module Magique module Backend class Redis def initialize(connection = ::Redis.new) @connection = connection end def push(job) @connection.lpush('magique:queue', JSON.dump(job)) end def shift _queue, job = @connection.brpop('magique:queue') payload = JSON.parse(job, symbolize_names: true) payload[:worker] = Object.const_get(payload[:worker]) payload end end end end

যেহেতু Redis রুবি অবজেক্ট সম্পর্কে কিছু জানে না, আমাদের টাস্কগুলো JSON-এ সিরিয়ালাইজ করতে হবে সেগুলো ডেটাবেসে স্টোর করার আগে lpush কমান্ড ব্যবহার করে যেটি লিস্টের সামনে এলিমেন্ট যোগ করে।

কিউ থেকে টাস্ক ফেচ করতে, আমরা brpop কমান্ড ব্যবহার করছি, যা লিস্ট থেকে লাস্ট এলিমেন্ট পায়। যদি লিস্ট খালি থাকে, এটি ব্লক করবে যতক্ষণ না নতুন এলিমেন্ট আসে। এটি আমাদের প্রসেসরগুলো পজ করার একটি সুন্দর পদ্ধতি যখন কোনো টাস্ক নেই। শেষে, Redis থেকে টাস্ক বের করার পর, আমাদের ওয়ার্কারের আসল রুবি ক্লাস লুকআপ করতে হবে নাম ব্যবহার করে Object.const_get দিয়ে।

শেষ ধাপ হিসেবে, আসুন জিনিসপত্র একাধিক প্রসেসে ভাগ করি। প্রডিউসার সাইডে, একমাত্র কাজ হলো ব্যাকএন্ডকে আমাদের নতুন ইমপ্লিমেন্ট করা Redis কিউতে পরিবর্তন করা।

# ... Magique.backend = Magique::Backend::Redis.new RUBYMAGIC.each do |url| TitleExtractorWorker.perform_async(url) end

কনসিউমার সাইডে, আমরা কয়েক লাইন দিয়ে বাঁচতে পারি এইভাবে:

# ... Magique.backend = Magique::Backend::Redis.new Magique::Processor.start(5) loop { sleep 1 }

এটি এক্সিকিউট করলে, কনসিউমার প্রসেস নতুন কাজের জন্য কিউতে অপেক্ষা করবে। একবার প্রডিউসার প্রসেস শুরু করলে যা টাস্ক কিউতে পুশ করে, আমরা দেখতে পাই সেগুলো তৎক্ষণাৎ প্রসেস হচ্ছে।

দায়িত্বের সাথে উপভোগ করুন এবং প্রোডাকশনে ব্যবহার করবেন না

যদিও আমরা এটিকে একটি রিয়েল ওয়ার্ল্ড সেটআপ থেকে দূরে রেখেছি যা আপনি প্রোডাকশনে ব্যবহার করবেন (তাই করবেন না!), আমরা একটি ব্যাকগ্রাউন্ড প্রসেসর বানাতে কিছু ধাপ নিলাম। আমরা একটি প্রসেসকে ব্যাকগ্রাউন্ড সার্ভিস হিসেবে চালানো শুরু করলাম। তারপর আমরা এটিকে অ্যাসিঙ্ক্রোনাস করলাম এবং Queue ব্যবহার করলাম প্রডিউসার-কনসিউমার সমস্যা সমাধান করতে। তারপর আমরা প্রসেসকে একাধিক প্রসেস বা মেশিনে বিস্তার করলাম Redis ব্যবহার করে ইন-মেমরি ইমপ্লিমেন্টেশনের বদলে।

যেমন আগে বললাম, এটি একটি সিম্পলিফাইড ইমপ্লিমেন্টেশন ব্যাকগ্রাউন্ড প্রসেসিং সিস্টেমের। অনেক কিছু মিসিং এবং এক্সপ্লিসিটলি ডিল্ট নয়। এগুলোর মধ্যে রয়েছে (এমনকি নন-লিমিটেড টু) এরর হ্যান্ডলিং, মাল্টিপল কিউ, শেডিউলিং, কানেকশন পুলিং, এবং সিগন্যাল হ্যান্ডলিং।

তবুও, আমরা লিখতে ফান করেছি এবং আশা করি আপনি একটি ব্যাকগ্রাউন্ড প্রসেসিং সিস্টেমের হুডের নিচে ঝাঁকি দিতে উপভোগ করেছেন। হয়তো আপনি এক বা دوটা জিনিস শিখেও গেছেন।