Merge pull request #278 from Yancey1989/fitaline_ft

Run fit-a-line demo with fault tolerant mode

Author: Yancey0623

demo/fit_a_line/train_ft.py

@@ -0,0 +1,66 @@
+import paddle.v2 as paddle
+import os
+import gzip
+from paddle.v2.reader.creator import cloud_reader
+import paddle.v2.dataset.uci_housing as uci_housing
+
+etcd_ip = os.getenv("ETCD_IP")
+etcd_endpoint = "http://" + etcd_ip + ":" + "2379"
+trainer_id = int(os.getenv("PADDLE_INIT_TRAINER_ID"))
+
+def main():
+    # init
+    paddle.init()
+
+    # network config
+    x = paddle.layer.data(name='x', type=paddle.data_type.dense_vector(13))
+    y_predict = paddle.layer.fc(input=x, size=1, act=paddle.activation.Linear())
+    y = paddle.layer.data(name='y', type=paddle.data_type.dense_vector(1))
+    cost = paddle.layer.mse_cost(input=y_predict, label=y)
+
+    # create parameters
+    parameters = paddle.parameters.create(cost)
+
+    # create optimizer
+    optimizer = paddle.optimizer.Momentum(momentum=0)
+
+    trainer = paddle.trainer.SGD(
+        cost=cost, 
+        parameters=parameters, 
+        update_equation=optimizer, 
+        is_local=False, 
+        pserver_spec=etcd_endpoint,
+        use_etcd=True)
+
+    feeding = {'x': 0, 'y': 1}
+
+    # event_handler to print training and testing info
+    def event_handler(event):
+        if isinstance(event, paddle.event.EndIteration):
+            if event.batch_id % 100 == 0:
+                print "Pass %d, Batch %d, Cost %f" % (
+                    event.pass_id, event.batch_id, event.cost)
+
+        if isinstance(event, paddle.event.EndPass):
+            result = trainer.test(
+                reader=paddle.batch(uci_housing.test(), batch_size=2),
+                feeding=feeding)
+            print "Test %d, Cost %f" % (event.pass_id, result.cost)
+            if trainer_id == "0":
+                with gzip.open("fit-a-line_pass_%05d.tar.gz" % event.pass_id,
+                               "w") as f:
+                    parameters.to_tar(f)
+    # training
+    trainer.train(
+        reader=paddle.batch(
+            paddle.reader.shuffle(cloud_reader(
+                ["/pfs/dlnel/public/dataset/uci_housing/uci_housing_train-*"],
+                etcd_endpoint), buf_size=500),
+            batch_size=2),
+        feeding=feeding,
+        event_handler=event_handler,
+        num_passes=30)
+
+
+if __name__ == '__main__':
+    main()

docker/k8s_tools.py

@@ -42,15 +42,10 @@ def fetch_pserver_ips():
     return ",".join(pserver_ips)
 
 def fetch_master_ip():
-    while True:
-        label_selector = "paddle-job-master=%s" % PADDLE_JOB_NAME
-        pod_list = fetch_pods_info(label_selector)
-        master_ip = ""
-        if len(pod_list) >=1:
-            master_ip = pod_list[0][1]
-        if master_ip:
-            return master_ip
-        time.sleep(5)
+    label_selector = "paddle-job-master=%s" % PADDLE_JOB_NAME
+    pod_list = fetch_pods_info(label_selector)
+    master_ips = [item[1] for item in pod_list]
+    return master_ips[0]
 
 def fetch_trainer_id():
     label_selector = "paddle-job=%s" % PADDLE_JOB_NAME

docker/paddle_k8s

@@ -12,12 +12,13 @@ start_pserver() {
 }
 
 start_new_pserver() {
+  stdbuf -oL python /root/k8s_tools.py wait_pods_running  paddle-job-master=${PADDLE_JOB_NAME} 1
   export MASTER_IP=$(python /root/k8s_tools.py fetch_master_ip)
   stdbuf -oL /usr/bin/pserver \
     -port=$PADDLE_INIT_PORT \
     -num-pservers=$PSERVERS \
     -log-level=debug \
-    -etcd-endpoint=http://$PADDLE_INIT_MASTER_IP:2379
+    -etcd-endpoint=http://$MASTER_IP:2379
 }
 
 start_master() {
@@ -43,6 +44,27 @@ check_trainer_ret() {
   exit $ret
 }
 
+start_new_trainer() {
+  stdbuf -oL python /root/k8s_tools.py wait_pods_running  paddle-job-master=${PADDLE_JOB_NAME} 1
+  # FIXME: use etcd lock instead of trainer id
+  stdbuf -oL python /root/k8s_tools.py wait_pods_running paddle-job=${PADDLE_JOB_NAME} ${TRAINERS} 
+
+  export MASTER_IP=$(python /root/k8s_tools.py fetch_master_ip)
+  export ETCD_IP="$MASTER_IP"
+  export PADDLE_INIT_TRAINER_ID=$(python /root/k8s_tools.py fetch_trainer_id)
+
+  # NOTE: $TRAINER_PACKAGE may be large, do not copy
+  export PYTHONPATH=$TRAINER_PACKAGE:$PYTHONPATH
+  cd $TRAINER_PACKAGE
+
+  stdbuf -oL echo "Starting training job: " $TRAINER_PACKAGE, "num_gradient_servers:" \
+  $PADDLE_INIT_NUM_GRADIENT_SERVERS, "trainer_id: " $PADDLE_INIT_TRAINER_ID, \
+  "version: " $1 
+
+  stdbuf -oL sh -c "${ENTRY}"
+  check_trainer_ret $?
+}
+
 start_trainer() {
     stdbuf -oL python /root/k8s_tools.py wait_pods_running paddle-job-pserver=${PADDLE_JOB_NAME} ${PSERVERS}
     stdbuf -oL python /root/k8s_tools.py wait_pods_running paddle-job=${PADDLE_JOB_NAME} ${TRAINERS}
@@ -98,6 +120,8 @@ usage() {
     echo "usage: paddle_k8s [<args>]:"
     echo "  start_trainer  [v1|v2]    Start a trainer process with v1 or v2 API"
     echo "  start_pserver             Start a pserver process"
+    echo "  start_new_pserver         Start a new pserver process"
+    echo "  start_new_trainer         Start a new triner process"
 }
 
 case "$1" in
@@ -107,6 +131,9 @@ case "$1" in
     start_trainer)
         start_trainer $2
         ;;
+    start_new_trainer)
+        start_new_trainer
+        ;;
     start_new_pserver)
         start_new_pserver
         ;;

paddlecloud/paddlejob/paddle_job.py

@@ -27,7 +27,7 @@ def __init__(self,
                  volumes=[],
                  registry_secret=None,
                  envs = {},
-                 new_pserver=True,
+                 fault_tolerant=False,
                  etcd_image="quay.io/coreos/etcd:v3.2.1"):
 
         self._ports_num=1
@@ -54,7 +54,7 @@ def __init__(self,
         self._mastercpu = 1
         self._mastermemory = "300Mi"
         # use new pserver for tolerant
-        self._new_pserver = new_pserver
+        self._fault_tolerant = fault_tolerant
         self._etcd_image = etcd_image
 
     @property
@@ -92,7 +92,6 @@ def get_env(self):
         envs.append({"name":"PADDLE_INIT_PORTS_NUM_FOR_SPARSE", "value":str(self._ports_num_for_sparse)})
         envs.append({"name":"PADDLE_INIT_NUM_GRADIENT_SERVERS", "value":str(self._num_gradient_servers)})
         envs.append({"name":"PADDLE_INIT_NUM_PASSES",           "value":str(self._passes)})
-
         if self._gpu:
             envs.append({"name":"PADDLE_INIT_USE_GPU", "value":str("1")})
             # HACK: add nvidia lib LD_LIBRARY_PATH for all pods
@@ -131,13 +130,15 @@ def _get_master_entrypoint(self):
         return ["paddle_k8s", "start_master"]
 
     def _get_pserver_entrypoint(self):
-        if not self._new_pserver:
+        if not self._fault_tolerant:
             return ["paddle_k8s", "start_pserver"]
         else:
             return ["paddle_k8s", "start_new_pserver"]
 
     def _get_trainer_entrypoint(self):
         if self._entry:
+            if self._fault_tolerant:
+                return ["paddle_k8s", "start_new_trainer"]
             return ["paddle_k8s", "start_trainer", "v2"]
         return ["paddle_k8s", "start_trainer", "v1"]
 

paddlecloud/paddlejob/views.py

@@ -146,7 +146,7 @@ def post(self, request, format=None):
             registry_secret = registry_secret,
             volumes = volumes,
             envs = envs,
-            new_pserver = fault_tolerant,
+            fault_tolerant = fault_tolerant,
             etcd_image = settings.ETCD_IMAGE
         )
         # ========== submit master ReplicaSet if using fault_tolerant feature ==