diff --git a/README.md b/README.md
index 2df2871..8a3166d 100644
--- a/README.md
+++ b/README.md
@@ -12,7 +12,7 @@ Implemented algorithms:
 * Async One-Step Sarsa 
 * Async N-Step Q-Learning
 * Continuous A3C
-* Deep Deterministic Policy Gradient (DDPG)
+* Distributed Deep Deterministic Policy Gradient (Distributed DDPG, aka D3PG)
 * Hybrid Reward Architecture (HRA)
 * Parallelized Proximal Policy Optimization (P3O, similar to DPPO)
 
@@ -47,15 +47,19 @@ For continuous A3C and DPPO, I use fixed unit variance rather than a separate he
 Of course you can also use another head to output variance. In that case, a good practice is to bound your mean while leave 
 variance unbounded, which is also included in the implementation.
 
-## DDPG
+## D3PG 
 
 ![Loading...](https://raw.githubusercontent.com/ShangtongZhang/DeepRL/master/images/DDPG.png)
 
 Extra caution is necessary when computing gradients. The [repo](https://github.com/ghliu/pytorch-ddpg) I referred
-is wrong in computing the deterministic gradients at least at this [commit](https://github.com/ghliu/pytorch-ddpg/tree/ffea335ee53f2ff90b6d7eaf9d0cee705270c0f1).
+for DDPG is wrong in computing the deterministic gradients at least at this [commit](https://github.com/ghliu/pytorch-ddpg/tree/ffea335ee53f2ff90b6d7eaf9d0cee705270c0f1).
 Theoretically I believe that implementation should work, but in practice it doesn't work. Even this is PyTorch you need to manually deal with gradients in this case.
 DDPG is not very stable. 
 
+Setting the number of workers to 1 will reduce the implementation to exact DDPG. I have to adopt the most straightforward distribution method, as
+P3O and A3C style distribution doesn't work for DDPG. The figures were done with 6 workers.
+
+
 ## P3O 
 
 ![Loading...](https://raw.githubusercontent.com/ShangtongZhang/DeepRL/master/images/P3O.png)
diff --git a/agent/DDPG_agent.py b/agent/DDPG_agent.py
deleted file mode 100644
index 07dc46e..0000000
--- a/agent/DDPG_agent.py
+++ /dev/null
@@ -1,104 +0,0 @@
-#######################################################################
-# Copyright (C) 2017 Shangtong Zhang(zhangshangtong.cpp@gmail.com)    #
-# Permission given to modify the code as long as you keep this        #
-# declaration at the top                                              #
-#######################################################################
-
-from network import *
-from component import *
-from utils import *
-import pickle
-import torch.nn as nn
-
-class DDPGAgent:
-    def __init__(self, config):
-        self.config = config
-        self.task = config.task_fn()
-        self.learning_network = config.network_fn()
-        self.target_network = config.network_fn()
-        self.target_network.load_state_dict(self.learning_network.state_dict())
-        self.target_network.eval()
-        self.actor_opt = config.actor_optimizer_fn(self.learning_network.actor.parameters())
-        self.critic_opt = config.critic_optimizer_fn(self.learning_network.critic.parameters())
-        self.replay = config.replay_fn()
-        self.random_process = config.random_process_fn()
-        self.criterion = nn.MSELoss()
-        self.total_steps = 0
-
-        self.state_normalizer = Normalizer(self.task.state_dim)
-        self.reward_normalizer = Normalizer(1)
-
-    def soft_update(self, target, src):
-        for target_param, param in zip(target.parameters(), src.parameters()):
-            target_param.data.copy_(target_param.data * (1.0 - self.config.target_network_mix) +
-                                    param.data * self.config.target_network_mix)
-
-    def episode(self, deterministic=False):
-        self.random_process.reset_states()
-        state = self.task.reset()
-        state = self.state_normalizer(state)
-
-        config = self.config
-        actor = self.learning_network.actor
-        critic = self.learning_network.critic
-        target_actor = self.target_network.actor
-        target_critic = self.target_network.critic
-
-        steps = 0
-        total_reward = 0.0
-        while True:
-            actor.eval()
-            action = actor.predict(np.stack([state])).flatten()
-            if not deterministic:
-                if self.total_steps < config.exploration_steps:
-                    action = self.task.random_action()
-                else:
-                    action += self.random_process.sample()
-            next_state, reward, done, info = self.task.step(action)
-            done = (done or (config.max_episode_length and steps >= config.max_episode_length))
-            next_state = self.state_normalizer(next_state)
-            total_reward += reward
-            # reward = self.reward_normalizer(reward)
-
-            if not deterministic:
-                self.replay.feed([state, action, reward, next_state, int(done)])
-                self.total_steps += 1
-            steps += 1
-            state = next_state
-
-            if done:
-                break
-
-            if not deterministic and self.total_steps > config.exploration_steps:
-                self.learning_network.train()
-                experiences = self.replay.sample()
-                states, actions, rewards, next_states, terminals = experiences
-                q_next = target_critic.predict(next_states, target_actor.predict(next_states))
-                terminals = critic.to_torch_variable(terminals).unsqueeze(1)
-                rewards = critic.to_torch_variable(rewards).unsqueeze(1)
-                q_next = config.discount * q_next * (1 - terminals)
-                q_next.add_(rewards)
-                q_next = q_next.detach()
-                q = critic.predict(states, actions)
-                critic_loss = self.criterion(q, q_next)
-
-                critic.zero_grad()
-                critic_loss.backward()
-                self.critic_opt.step()
-
-                actions = actor.predict(states, False)
-                var_actions = Variable(actions.data, requires_grad=True)
-                q = critic.predict(states, var_actions)
-                q.backward(torch.ones(q.size()))
-
-                actor.zero_grad()
-                actions.backward(-var_actions.grad.data)
-                self.actor_opt.step()
-
-                self.soft_update(self.target_network, self.learning_network)
-
-        return total_reward, steps
-
-    def save(self, file_name):
-        with open(file_name, 'wb') as f:
-            pickle.dump(self.learning_network.state_dict(), f)
diff --git a/agent/__init__.py b/agent/__init__.py
index b77c3eb..8462c94 100644
--- a/agent/__init__.py
+++ b/agent/__init__.py
@@ -1,3 +1,2 @@
 from .async_agent import *
-from .DDPG_agent import *
 from .DQN_agent import *
diff --git a/agent/async_agent.py b/agent/async_agent.py
index c0fdac4..7bc12f8 100644
--- a/agent/async_agent.py
+++ b/agent/async_agent.py
@@ -51,7 +51,7 @@ def evaluate(config, task, learning_network, extra):
             with open('data/%s-%s-statistics-%s.bin' % (
                     config.tag, config.worker.__name__, task.name), 'wb') as f:
                 pickle.dump([test_rewards, test_points, test_wall_times], f)
-            if np.mean(rewards) > task.success_threshold or (config.max_steps and steps >= config.max_steps):
+            if np.mean(rewards) >= config.success_threshold or (config.max_steps and steps >= config.max_steps):
                 config.stop_signal.value = True
                 break
 
@@ -87,7 +87,6 @@ class AsyncAgent:
             extra = None
         args = [(i, config, learning_network, extra) for i in range(config.num_workers)]
         args.append((config, task, learning_network, extra))
-        # procs = []
         procs = [mp.Process(target=evaluate, args=args[-1])]
         procs.extend([mp.Process(target=train, args=args[i]) for i in range(config.num_workers)])
         for p in procs: p.start()
diff --git a/async_worker/actor_critic.py b/async_worker/actor_critic.py
index 3ec71a7..9ad9417 100644
--- a/async_worker/actor_critic.py
+++ b/async_worker/actor_critic.py
@@ -7,6 +7,7 @@ import numpy as np
 import torch
 from torch.autograd import Variable
 import torch.nn as nn
+from utils import *
 
 class AdvantageActorCritic:
     def __init__(self, config, learning_network, target_network):
@@ -68,11 +69,7 @@ class AdvantageActorCritic:
                 self.optimizer.zero_grad()
                 loss.backward()
                 nn.utils.clip_grad_norm(self.worker_network.parameters(), config.gradient_clip)
-                for param, worker_param in zip(
-                        self.learning_network.parameters(), self.worker_network.parameters()):
-                    if param.grad is not None:
-                        break
-                    param._grad = worker_param.grad
+                sync_grad(self.learning_network, self.worker_network)
                 self.optimizer.step()
                 self.worker_network.load_state_dict(self.learning_network.state_dict())
                 self.worker_network.reset(terminal)
diff --git a/async_worker/continuous_actor_critic.py b/async_worker/continuous_actor_critic.py
index dabb0dd..31b8abb 100644
--- a/async_worker/continuous_actor_critic.py
+++ b/async_worker/continuous_actor_critic.py
@@ -92,11 +92,7 @@ class ContinuousAdvantageActorCritic:
                 actor_loss.backward()
                 critic_loss.backward()
                 nn.utils.clip_grad_norm(self.worker_network.parameters(), config.gradient_clip)
-                for param, worker_param in zip(
-                        self.learning_network.parameters(), self.worker_network.parameters()):
-                    if param.grad is not None:
-                        break
-                    param._grad = worker_param.grad
+                sync_grad(self.learning_network, self.worker_network)
                 self.actor_opt.step()
                 self.critic_opt.step()
                 self.worker_network.load_state_dict(self.learning_network.state_dict())
diff --git a/async_worker/dpg.py b/async_worker/dpg.py
index bb74296..6137c25 100644
--- a/async_worker/dpg.py
+++ b/async_worker/dpg.py
@@ -30,11 +30,9 @@ class DeterministicPolicyGradient:
         self.random_process = config.random_process_fn()
         self.criterion = nn.MSELoss()
 
-        # self.state_normalizer = Normalizer(self.task.state_dim)
-        self.shared_state_normalizer = extra[0]
+        self.shared_state_normalizer, self.shared_reward_normalizer, self.replay = extra
         self.state_normalizer = StaticNormalizer(self.task.state_dim)
-        # self.replay = config.replay_fn()
-        self.replay = extra[-1]
+        self.reward_normalizer = StaticNormalizer(1)
 
     def soft_update(self, target, src):
         for target_param, param in zip(target.parameters(), src.parameters()):
@@ -63,6 +61,7 @@ class DeterministicPolicyGradient:
             done = (done or (config.max_episode_length and steps >= config.max_episode_length))
             next_state = self.state_normalizer(next_state)
             total_reward += reward
+            reward = self.reward_normalizer(reward)
 
             if not deterministic:
                 self.replay.feed([state, action, reward, next_state, int(done)])
@@ -92,10 +91,7 @@ class DeterministicPolicyGradient:
                 self.critic_opt.zero_grad()
                 critic_loss.backward()
                 with config.network_lock:
-                    for param, worker_param in zip(self.shared_network.critic.parameters(), critic.parameters()):
-                        if param.grad is not None:
-                            break
-                        param._grad = worker_param.grad
+                    sync_grad(self.shared_network.critic, critic)
                     self.critic_opt.step()
 
                 actions = actor.predict(states, False)
@@ -107,14 +103,17 @@ class DeterministicPolicyGradient:
                 self.actor_opt.zero_grad()
                 actions.backward(-var_actions.grad.data)
                 with config.network_lock:
-                    for param, worker_param in zip(self.shared_network.actor.parameters(), actor.parameters()):
-                        if param.grad is not None:
-                            break
-                        param._grad = worker_param.grad
+                    sync_grad(self.shared_network.actor, actor)
                     self.actor_opt.step()
 
                 self.worker_network.load_state_dict(self.shared_network.state_dict())
 
                 self.soft_update(self.target_network, self.worker_network)
 
+        self.shared_state_normalizer.offline_stats.merge(self.state_normalizer.online_stats)
+        self.state_normalizer.online_stats.zero()
+
+        self.shared_reward_normalizer.offline_stats.merge(self.reward_normalizer.online_stats)
+        self.reward_normalizer.online_stats.zero()
+
         return steps, total_reward
diff --git a/async_worker/n_step_q.py b/async_worker/n_step_q.py
index 3a858cf..3a5899b 100644
--- a/async_worker/n_step_q.py
+++ b/async_worker/n_step_q.py
@@ -7,6 +7,7 @@ import numpy as np
 import torch
 from torch.autograd import Variable
 import torch.nn as nn
+from utils import *
 
 class NStepQLearning:
     def __init__(self, config, learning_network, target_network):
@@ -63,11 +64,7 @@ class NStepQLearning:
                 self.optimizer.zero_grad()
                 loss.backward()
                 nn.utils.clip_grad_norm(self.worker_network.parameters(), config.gradient_clip)
-                for param, worker_param in zip(
-                        self.learning_network.parameters(), self.worker_network.parameters()):
-                    if param.grad is not None:
-                        break
-                    param._grad = worker_param.grad
+                sync_grad(self.learning_network, self.worker_network)
                 self.optimizer.step()
                 self.worker_network.load_state_dict(self.learning_network.state_dict())
                 self.worker_network.reset(terminal)
diff --git a/async_worker/one_step_q.py b/async_worker/one_step_q.py
index 9f1213e..d190753 100644
--- a/async_worker/one_step_q.py
+++ b/async_worker/one_step_q.py
@@ -7,6 +7,7 @@ import numpy as np
 import torch
 from torch.autograd import Variable
 import torch.nn as nn
+from utils import *
 
 class OneStepQLearning:
     def __init__(self, config, learning_network, target_network):
@@ -60,11 +61,7 @@ class OneStepQLearning:
                 self.optimizer.zero_grad()
                 loss.backward()
                 nn.utils.clip_grad_norm(self.worker_network.parameters(), config.gradient_clip)
-                for param, worker_param in zip(
-                        self.learning_network.parameters(), self.worker_network.parameters()):
-                    if param.grad is not None:
-                        break
-                    param._grad = worker_param.grad
+                sync_grad(self.learning_network, self.worker_network)
                 self.optimizer.step()
                 self.worker_network.load_state_dict(self.learning_network.state_dict())
                 self.worker_network.reset(terminal)
diff --git a/async_worker/one_step_sarsa.py b/async_worker/one_step_sarsa.py
index b5123e9..d763867 100644
--- a/async_worker/one_step_sarsa.py
+++ b/async_worker/one_step_sarsa.py
@@ -7,6 +7,7 @@ import numpy as np
 import torch
 from torch.autograd import Variable
 import torch.nn as nn
+from utils import *
 
 class OneStepSarsa:
     def __init__(self, config, learning_network, target_network):
@@ -65,11 +66,7 @@ class OneStepSarsa:
                 self.optimizer.zero_grad()
                 loss.backward()
                 nn.utils.clip_grad_norm(self.worker_network.parameters(), config.gradient_clip)
-                for param, worker_param in zip(
-                        self.learning_network.parameters(), self.worker_network.parameters()):
-                    if param.grad is not None:
-                        break
-                    param._grad = worker_param.grad
+                sync_grad(self.learning_network, self.worker_network)
                 self.optimizer.step()
                 self.worker_network.load_state_dict(self.learning_network.state_dict())
                 self.worker_network.reset(terminal)
diff --git a/async_worker/ppo.py b/async_worker/ppo.py
index b0992ee..58f1ea8 100644
--- a/async_worker/ppo.py
+++ b/async_worker/ppo.py
@@ -152,8 +152,7 @@ class ProximalPolicyOptimization:
                 self.shared_network.zero_grad()
                 self.actor_opt.zero_grad()
                 self.critic_opt.zero_grad()
-                for param, worker_param in zip(self.shared_network.parameters(), self.worker_network.parameters()):
-                    param._grad = worker_param.grad.clone()
+                sync_grad(self.shared_network, self.worker_network)
                 self.actor_opt.step()
                 self.critic_opt.step()
 
diff --git a/images/DDPG-RoboschoolReacher-v1.png b/images/DDPG-RoboschoolReacher-v1.png
deleted file mode 100644
index 08d937c..0000000
Binary files a/images/DDPG-RoboschoolReacher-v1.png and /dev/null differ
diff --git a/images/DDPG.png b/images/DDPG.png
index 2841c5d..164511c 100644
Binary files a/images/DDPG.png and b/images/DDPG.png differ
diff --git a/main.py b/main.py
index 06aa69a..0ecda45 100644
--- a/main.py
+++ b/main.py
@@ -201,9 +201,10 @@ def a3c_continuous():
 
 def p3o_continuous():
     config = Config()
-    # config.task_fn = lambda: Pendulum()
+    config.task_fn = lambda: Pendulum()
+    config.task_fn = lambda: BipedalWalker()
     # config.task_fn = lambda: BipedalWalkerHardcore()
-    config.task_fn = lambda: Roboschool('RoboschoolInvertedPendulum-v1')
+    # config.task_fn = lambda: Roboschool('RoboschoolInvertedPendulum-v1')
     # config.task_fn = lambda: Roboschool('RoboschoolAnt-v1')
     task = config.task_fn()
     config.actor_network_fn = lambda: GaussianActorNet(task.state_dim, task.action_dim,
@@ -214,60 +215,29 @@ def p3o_continuous():
     config.critic_optimizer_fn = lambda params: torch.optim.Adam(params, 0.001)
 
     config.policy_fn = lambda: GaussianPolicy()
-    # config.replay_fn = lambda: GeneralReplay(memory_size=2048, batch_size=2048)
-    config.replay_fn = lambda: GeneralReplay(memory_size=2048, batch_size=64)
+    config.replay_fn = lambda: GeneralReplay(memory_size=2048, batch_size=2048)
     config.worker = ProximalPolicyOptimization
     config.discount = 0.99
     config.gae_tau = 0.97
-    config.num_workers = 8
+    config.num_workers = 6
     config.test_interval = 1
     config.test_repetitions = 1
     config.max_episode_length = task.max_episode_steps
     config.entropy_weight = 0
     config.gradient_clip = 20
     config.rollout_length = 10000
-    config.optimize_epochs = 10
+    config.optimize_epochs = 1
     config.ppo_ratio_clip = 0.2
     config.logger = Logger('./log', gym.logger)
     agent = AsyncAgent(config)
     agent.run()
 
-def ddpg_continuous():
-    config = Config()
-    # config.task_fn = lambda: Pendulum()
-    # config.task_fn = lambda: BipedalWalker()
-    config.task_fn = lambda: ContinuousLunarLander()
-    # config.task_fn = lambda: Roboschool('RoboschoolInvertedPendulum-v1')
-    # config.task_fn = lambda: Roboschool('RoboschoolReacher-v1')
-    task = config.task_fn()
-    config.actor_network_fn = lambda: DeterministicActorNet(
-        task.state_dim, task.action_dim, F.tanh, 2, non_linear=F.relu, batch_norm=False)
-    config.critic_network_fn = lambda: DeterministicCriticNet(
-        task.state_dim, task.action_dim, non_linear=F.relu, batch_norm=False)
-    config.network_fn = lambda: DisjointActorCriticNet(config.actor_network_fn, config.critic_network_fn)
-    config.actor_optimizer_fn = lambda params: torch.optim.Adam(params, lr=1e-4)
-    config.critic_optimizer_fn =\
-        lambda params: torch.optim.Adam(params, lr=1e-3, weight_decay=0.01)
-    config.replay_fn = lambda: HighDimActionReplay(memory_size=1000000, batch_size=64)
-    config.discount = 0.99
-    config.max_episode_length = task.max_episode_steps
-    config.target_network_mix = 0.001
-    config.exploration_steps = 100
-    config.random_process_fn = \
-        lambda: OrnsteinUhlenbeckProcess(size=task.action_dim, theta=0.15, sigma=0.2,
-                                         n_steps_annealing=10000)
-    config.test_interval = 0
-    config.test_repetitions = 10
-    config.save_interval = 50
-    config.logger = Logger('./log', gym.logger)
-    run_episodes(DDPGAgent(config))
-
-def addpg_continuous():
+def d3pg_continuous():
     config = Config()
     # config.task_fn = lambda: Pendulum()
     # config.task_fn = lambda: ContinuousLunarLander()
-    config.task_fn = lambda: Roboschool('RoboschoolInvertedPendulum-v1')
-    # config.task_fn = lambda: Roboschool('RoboschoolReacher-v1')
+    # config.task_fn = lambda: Roboschool('RoboschoolInvertedPendulum-v1')
+    config.task_fn = lambda: Roboschool('RoboschoolReacher-v1')
     # config.task_fn = lambda: BipedalWalker()
     task = config.task_fn()
     config.actor_network_fn = lambda: DeterministicActorNet(
@@ -278,10 +248,8 @@ def addpg_continuous():
     config.actor_optimizer_fn = lambda params: torch.optim.Adam(params, lr=1e-4)
     config.critic_optimizer_fn =\
         lambda params: torch.optim.Adam(params, lr=1e-4)
-    # config.replay_fn = lambda: HighDimActionReplay(memory_size=1000000, batch_size=64)
     config.replay_fn = lambda: SharedReplay(memory_size=1000000, batch_size=64,
                                             state_shape=(task.state_dim, ), action_shape=(task.action_dim, ))
-    # config.replay_fn = lambda: GeneralReplay(memory_size=256, batch_size=64)
     config.discount = 0.99
     config.max_episode_length = task.max_episode_steps
     config.random_process_fn = \
@@ -291,27 +259,23 @@ def addpg_continuous():
     config.num_workers = 6
     config.min_memory_size = 50
     config.target_network_mix = 0.001
-    # config.update_interval = 10
     config.test_interval = 500
     config.test_repetitions = 1
     config.gradient_clip = 20
-    config.rollout_length = 16
-    config.optimize_epochs = 1
     config.logger = Logger('./log', gym.logger)
     agent = AsyncAgent(config)
     agent.run()
 
 if __name__ == '__main__':
-    gym.logger.setLevel(logging.DEBUG)
-    # gym.logger.setLevel(logging.INFO)
+    # gym.logger.setLevel(logging.DEBUG)
+    gym.logger.setLevel(logging.INFO)
 
     # dqn_cart_pole()
     # async_cart_pole()
     # a3c_cart_pole()
     # a3c_continuous()
     # p3o_continuous()
-    # ddpg_continuous()
-    addpg_continuous()
+    d3pg_continuous()
 
     # dqn_fruit()
     # hrdqn_fruit()
diff --git a/utils/__init__.py b/utils/__init__.py
index ba7d6a7..c9b2200 100644
--- a/utils/__init__.py
+++ b/utils/__init__.py
@@ -1,6 +1,6 @@
 from .config import *
 from .normalizer import *
-from .run import *
+from .misc import *
 
 try:
     from .tf_logger import Logger
diff --git a/utils/config.py b/utils/config.py
index bad7273..132a866 100644
--- a/utils/config.py
+++ b/utils/config.py
@@ -48,3 +48,4 @@ class Config:
         self.min_epsilon = 0
         self.save_interval = 0
         self.max_steps = 0
+        self.success_threshold = float('inf')
diff --git a/utils/run.py b/utils/misc.py
similarity index 90%
rename from utils/run.py
rename to utils/misc.py
index 3976404..8f4ba1a 100644
--- a/utils/run.py
+++ b/utils/misc.py
@@ -50,4 +50,8 @@ def run_episodes(agent):
             if avg_reward > agent.task.success_threshold:
                 break
 
-    return steps, rewards, avg_test_rewards
\ No newline at end of file
+    return steps, rewards, avg_test_rewards
+
+def sync_grad(target_network, src_network):
+    for param, src_param in zip(target_network.parameters(), src_network.parameters()):
+        param._grad = src_param.grad.clone()
\ No newline at end of file