Early stopping with ddp  bug
Describe the bug
Earley stopping with ddp stalls :
When using distribued mode ddp and early stopping if the stop condition is met in one or more subprocess but not in all subprocess, the corresponding subprocess are stop but the others ones are still running and the training hangs.
with dp mode all is doing fine
To Reproduce
I use the example code, with a forked early stopping callback wich stops if val_acc>threshold
I also fix the bug #371 in trainer.py  line 1131 in met_batch_limit = batch_nb > self.nb_training_batches  (i.e change > by >= )
In validation_end() i return :
result = {'progress_bar': tqdm_dict, 'log': tqdm_dict,'val_acc':val_acc_mean}
return result
I use pytorch 1.2
The code :
from pytorch_lightning.callbacks import EarlyStopping

class ThresholdStopping(EarlyStopping):
   
    def __init__(self, monitor="val_acc",thresh=0.0,mode="max",verbose=0):
        super(ThresholdStopping, self).__init__()

        self.monitor = monitor
        self.stopped_epoch = 0
        self.thresh=thresh
        self.verbose = verbose

        if mode not in ['auto', 'min', 'max']:
            print('EarlyStopping mode %s is unknown, fallback to auto mode.' % mode)
            mode = 'auto'

        if mode == 'min':
            self.monitor_op = np.less
        elif mode == 'max':
            self.monitor_op = np.greater
        else:
            if 'acc' in self.monitor:
                self.monitor_op = np.greater
            else:
                self.monitor_op = np.less

        self.on_train_begin()

    def on_train_begin(self, logs=None):
        self.stopped_epoch = 0

    def on_epoch_end(self, epoch, logs=None):
        
        current = logs.get(self.monitor)
        stop_training = False

        print("current",current,self.thresh)
        
        if current is None:
            print('Early stopping conditioned on metric `%s` '
                  'which is not available. Available metrics are: %s' %
                  (self.monitor, ','.join(list(logs.keys()))), RuntimeWarning)
            stop_training = True
            return stop_training

        if self.monitor_op(current ,self.thresh):
            
            self.stopped_epoch = epoch
            stop_training = True
            self.on_train_end()

        return stop_training

    def on_train_end(self, logs=None):
        if self.stopped_epoch >= 0:
            print('Epoch %05d: early stopping' % (self.stopped_epoch + 1))

def main(hparams):
    """
    Main training routine specific for this project
    :param hparams:
    """
    # ------------------------
    # 1 INIT LIGHTNING MODEL
    # ------------------------
    model = LightningTemplateModel(hparams)

    # ------------------------
    # 2 INIT TRAINER
    # ------------------------
    # trainer = Trainer(max_nb_epochs=1, gpus=[0, 1, 3, 7], distributed_backend='ddp')   

    gpu_param=hparams.gpus
    
    early_stop_callback=ThresholdStopping(monitor='val_acc',thresh=0.85)

    if hparams.distributed_backend=="ddp":
        gpu_param=[0,1,2,3,4, 5, 6, 7]

    trainer = Trainer(
        gpus=gpu_param,
        distributed_backend=hparams.distributed_backend,
        use_amp=hparams.use_16bit,
        early_stop_callback=early_stop_callback,
        min_nb_epochs=-1,
        max_nb_epochs=20,
    )

    # ------------------------
    # 3 START TRAINING
    # ------------------------
    trainer.fit(model)

if __name__ == '__main__':
    # ------------------------
    # TRAINING ARGUMENTS
    # ------------------------
    # these are project-wide arguments

    root_dir = os.path.dirname(os.path.realpath(__file__))
    parent_parser = ArgumentParser(add_help=False)

    # gpu args
    parent_parser.add_argument(
        '--gpus',
        type=int,
        default=2,
        help='how many gpus'
    )
    parent_parser.add_argument(
        '--distributed_backend',
        type=str,
        default='dp',
        help='supports three options dp, ddp, ddp2'
    )
    parent_parser.add_argument(
        '--use_16bit',
        dest='use_16bit',
        action='store_true',
        help='if true uses 16 bit precision'
    )

    # each LightningModule defines arguments relevant to it
    parser = LightningTemplateModel.add_model_specific_args(parent_parser, root_dir)
    hyperparams = parser.parse_args()

    # ---------------------
    # RUN TRAINING
    # ---------------------
    main(hyperparams)

Expected behavior
All the subprocess should stop
Screenshots
If applicable, add screenshots to help explain your problem.

Desktop (please complete the following information):

OS:  Ubuntu 16.04.4 LTS
Browser [e.g. chrome, safari]
Version : pip version

Additional context