Replace jar distribution strategy with bittorrent #629
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -9,6 +9,7 @@ | |
| (:use [backtype.storm.scheduler.DefaultScheduler]) | ||
| (:import [backtype.storm.scheduler INimbus SupervisorDetails WorkerSlot TopologyDetails | ||
| Cluster Topologies SchedulerAssignment SchedulerAssignmentImpl DefaultScheduler ExecutorDetails]) | ||
| (:import [backtype.storm.torrent NimbusTracker]) | ||
| (:use [backtype.storm bootstrap util]) | ||
| (:use [backtype.storm.daemon common]) | ||
| (:gen-class | ||
|
|
@@ -43,6 +44,8 @@ | |
| scheduler | ||
| )) | ||
|
|
||
| (defmulti mk-bt-tracker cluster-mode) | ||
|
|
||
| (defn nimbus-data [conf inimbus] | ||
| (let [forced-scheduler (.getForcedScheduler inimbus)] | ||
| {:conf conf | ||
|
|
@@ -60,6 +63,7 @@ | |
| (halt-process! 20 "Error when processing an event") | ||
| )) | ||
| :scheduler (mk-scheduler conf inimbus) | ||
| :bt-tracker (mk-bt-tracker conf) | ||
| })) | ||
|
|
||
| (defn inbox [nimbus] | ||
|
|
@@ -290,13 +294,14 @@ | |
| ;; need to somehow maintain stream/component ids inside tuples | ||
| topology) | ||
|
|
||
| (defn- setup-storm-code [nimbus conf storm-id tmp-jar-location storm-conf topology] | ||
| (let [stormroot (master-stormdist-root conf storm-id)] | ||
| (FileUtils/forceMkdir (File. stormroot)) | ||
| (FileUtils/cleanDirectory (File. stormroot)) | ||
| (setup-jar conf tmp-jar-location stormroot) | ||
| (FileUtils/writeByteArrayToFile (File. (master-stormcode-path stormroot)) (Utils/serialize topology)) | ||
| (FileUtils/writeByteArrayToFile (File. (master-stormconf-path stormroot)) (Utils/serialize storm-conf)) | ||
| (if (:bt-tracker nimbus) (.trackAndSeed (:bt-tracker nimbus) stormroot storm-id)) | ||
| )) | ||
|
|
||
| (defn- read-storm-topology [conf storm-id] | ||
|
|
@@ -812,6 +817,7 @@ | |
| (when-not (empty? to-cleanup-ids) | ||
| (doseq [id to-cleanup-ids] | ||
| (log-message "Cleaning up " id) | ||
| (if (:bt-tracker nimbus) (.stop (:bt-tracker nimbus) id)) | ||
| (.teardown-heartbeats! storm-cluster-state id) | ||
| (.teardown-topology-errors! storm-cluster-state id) | ||
| (rmr (master-stormdist-root conf id)) | ||
|
|
@@ -928,7 +934,7 @@ | |
| ;; lock protects against multiple topologies being submitted at once and | ||
| ;; cleanup thread killing topology in b/w assignment and starting the topology | ||
| (locking (:submit-lock nimbus) | ||
| (setup-storm-code nimbus conf storm-id uploadedJarLocation storm-conf topology) | ||
| (.setup-heartbeats! storm-cluster-state storm-id) | ||
| (let [thrift-status->kw-status {TopologyInitialStatus/INACTIVE :inactive | ||
| TopologyInitialStatus/ACTIVE :active}] | ||
|
|
@@ -1151,12 +1157,21 @@ | |
| (FileUtils/copyFile src-file (File. (master-stormjar-path stormroot))) | ||
| )) | ||
|
|
||
| (defmethod mk-bt-tracker :distributed [conf] | ||
|
There was a problem hiding this comment. Do we require the use of bit-torrent distribution? It will be better if we could make it optional. For simple topology, one may just prefer not to use BT at all. There was a problem hiding this comment. Yes it is required. There was some discussion around making the distribution mechanism pluggable, but Nathan argued against it, as making it pluggable would increase complexity (#629 (comment)). |
||
| (NimbusTracker. conf) | ||
| ) | ||
|
|
||
|
|
||
| ;; local implementation | ||
|
|
||
| (defmethod setup-jar :local [conf & args] | ||
| nil | ||
| ) | ||
|
|
||
| (defmethod mk-bt-tracker :local [conf] | ||
| nil | ||
| ) | ||
|
|
||
| (defn -launch [nimbus] | ||
| (launch-server! (read-storm-config) nimbus)) | ||
|
|
||
|
|
||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,34 @@ | ||
| package backtype.storm.torrent; | ||
|
|
||
| import java.util.HashMap; | ||
|
|
||
| import org.slf4j.Logger; | ||
| import org.slf4j.LoggerFactory; | ||
|
|
||
| import com.turn.ttorrent.client.Client; | ||
|
|
||
| public abstract class BaseTracker { | ||
|
There was a problem hiding this comment. I'm not sure that you want to call it as BaseTracker. NimbusTracker seems to make sense, but SupervisorTracker is not. From BT point of view, we only have 1 tracker located on Nimbus box. There was a problem hiding this comment. I'll admit that the "Tracker" name is a bit overloaded here, and I'm not tied to any naming convention. I re-used the "Tracker" name because both the SupervisorTracker and NimbusTracker classes have to "keep track of" BT clients. I'm perfectly open to renaming those classes. |
||
| private static final Logger LOG = LoggerFactory.getLogger(BaseTracker.class); | ||
|
|
||
| protected HashMap<String, Client> clients = new HashMap<String, Client>(); | ||
| protected Double maxDownload; | ||
| protected Double maxUpload; | ||
|
|
||
| protected synchronized void rebalanceRates(){ | ||
| int clientCount = this.clients.size(); | ||
| if(clientCount > 0){ | ||
| double maxDl = this.maxDownload <= 0.0 ? this.maxDownload : this.maxDownload / clientCount; | ||
| double maxUl = this.maxUpload <= 0.0 ? this.maxUpload : this.maxUpload / clientCount; | ||
| LOG.info("Rebalancing bandwidth allocation based on {} topology torrents.", clientCount); | ||
| LOG.info("Per-torrent allocation [D/U]: {}/{} kB/sec.", format(maxDl), format(maxUl)); | ||
| for(Client client : this.clients.values()) { | ||
| client.setMaxDownloadRate(maxDl); | ||
| client.setMaxUploadRate(maxUl); | ||
| } | ||
| } | ||
| } | ||
|
|
||
| protected static String format(double val){ | ||
| return val <= 0.0 ? "UNLIMITED" : String.format("%.2f", val); | ||
| } | ||
| } | ||
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
0.0 means no limit?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Ah, after read the comments in
Config.java, I know that0.0meansunlimited, do we want to give user a more practical default?There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
It seemed to me that throttling would be enabled for special cases -- big cluster, large topology, etc. I would imagine for most cases we would not want to throttle.