text:
module Rack
  module OAuth2
    class Server

      class Client

        class << self
          # Authenticate a client request. This method takes three arguments,
          # Find Client from client identifier.
          def find(client_id)
            id = Server.options[:pk_generator].from_string(client_id.to_s)
            Server.new_instance self, collection.find_one({ :_id=>id })
          rescue BSON::InvalidObjectId, InvalidUUID
          end

          # Create a new client. Client provides the following properties:
          # # :display_name -- Name to show (e.g. UberClient)
          # # :link -- Link to client Web site (e.g. http://uberclient.dot)
          # # :image_url -- URL of image to show alongside display name
          # # :redirect_uri -- Registered redirect URI.
          # # :scope -- List of names the client is allowed to request.
          # # :notes -- Free form text.
          #
          # This method does not validate any of these fields, in fact, you're
          # not required to set them, use them, or use them as suggested. Using
          # them as suggested would result in better user experience.  Don't ask
          # how we learned that.
          def create(args)
            redirect_uri = Server::Utils.parse_redirect_uri(args[:redirect_uri]).to_s if args[:redirect_uri]
            scope = Server::Utils.normalize_scope(args[:scope])
            fields =  { :display_name=>args[:display_name], :link=>args[:link],
                        :image_url=>args[:image_url], :redirect_uri=>redirect_uri,
                        :notes=>args[:notes].to_s, :scope=>scope,
                        :created_at=>Time.now.to_i, :revoked=>nil }
            if args[:id] && args[:secret]
              fields[:_id], fields[:secret] = Server.options[:pk_generator].from_string(args[:id].to_s), args[:secret]
              collection.insert(fields, :safe=>true)
            else
              fields[:secret] = Server.secure_random
              fields[:_id] = Server.options[:pk_generator].generate
              collection.insert(fields)
            end
            Server.new_instance self, fields
          end

          # Lookup client by ID, display name or URL.
          def lookup(field)
            id = Server.options[:pk_generator].from_string(field.to_s)
            Server.new_instance self, collection.find_one(id)
          rescue BSON::InvalidObjectId, InvalidUUID
            Server.new_instance self, collection.find_one({ :display_name=>field }) || collection.find_one({ :link=>field })
          end

          # Returns all the clients in the database, sorted alphabetically.
          def all
            collection.find({}, { :sort=>[[:display_name, Mongo::ASCENDING]] }).
              map { |fields| Server.new_instance self, fields }
          end

          # Deletes client with given identifier (also, all related records).
          def delete(client_id)
            id = Server.options[:pk_generator].from_string(client_id.to_s)
            Client.collection.remove({ :_id=>id })
            AuthRequest.collection.remove({ :client_id=>id })
            AccessGrant.collection.remove({ :client_id=>id })
            AccessToken.collection.remove({ :client_id=>id })
          end

          def collection
            prefix = Server.options[:collection_prefix]
            Server.database["#{prefix}.clients"]
          end
        end

        # Client identifier.
        attr_reader :_id
        alias :id :_id
        # Client secret: random, long, and hexy.
        attr_reader :secret
        # User see this.
        attr_reader :display_name
        # Link to client's Web site.
        attr_reader :link
        # Preferred image URL for this icon.
        attr_reader :image_url
        # Redirect URL. Supplied by the client if they want to restrict redirect
        # URLs (better security).
        attr_reader :redirect_uri
        # List of scope the client is allowed to request.
        attr_reader :scope
        # Free form fields for internal use.
        attr_reader :notes
        # Does what it says on the label.
        attr_reader :created_at
        # Timestamp if revoked.
        attr_accessor :revoked
        # Counts how many access tokens were granted.
        attr_reader :tokens_granted
        # Counts how many access tokens were revoked.
        attr_reader :tokens_revoked

        # Revoke all authorization requests, access grants and access tokens for
        # this client. Ward off the evil.
        def revoke!
          self.revoked = Time.now.to_i
          Client.collection.update({ :_id=>id }, { :$set=>{ :revoked=>revoked } })
          AuthRequest.collection.update({ :client_id=>id }, { :$set=>{ :revoked=>revoked } })
          AccessGrant.collection.update({ :client_id=>id }, { :$set=>{ :revoked=>revoked } })
          AccessToken.collection.update({ :client_id=>id }, { :$set=>{ :revoked=>revoked } })
        end

        def update(args)
          fields = [:display_name, :link, :image_url, :notes].inject({}) { |h,k| v = args[k]; h[k] = v if v; h }
          fields[:redirect_uri] = Server::Utils.parse_redirect_uri(args[:redirect_uri]).to_s if args[:redirect_uri]
          fields[:scope] = Server::Utils.normalize_scope(args[:scope])
          self.class.collection.update({ :_id=>id }, { :$set=>fields })
          self.class.find(id)
        end

        Server.create_indexes do
          # For quickly returning clients sorted by display name, or finding
          # client from a URL.
          collection.create_index [[:display_name, Mongo::ASCENDING]]
          collection.create_index [[:link, Mongo::ASCENDING]]
        end
      end

    end
  end
end


text:
package org.softeg.slartus.forpdaplus.listtemplates;/*
 * Created by slinkin on 17.03.14.
 */

import android.support.v4.app.Fragment;

import org.softeg.slartus.forpdaplus.listfragments.AppsGamesCatalogFragment;
import org.softeg.slartus.forpdaplus.listfragments.AppsGamesTopicsListFragment;

public class AppsGamesTopicsBrickInfo extends BrickInfo {

    @Override
    public String getTitle() {
        return "Каталог@темы";
    }

    @Override
    public String getName() {
        return "AppsGamesTopics";
    }

    @Override
    public Fragment createFragment() {
        return new AppsGamesTopicsListFragment().setBrickInfo(this);
    }
}

redpajama_set_name: RedPajamaGithub


text:
<TITLE>Review for Your Friends & Neighbors (1998)</TITLE>
<LINK REL="STYLESHEET" TYPE="text/css" HREF="/ramr.css">

Your Friends & Neighbors (1998)

reviewed by
Murali Krishnan

Your Friends and Neighbors

[2.5/4.0]

Jerry is a drama professor and his wife, Terri, is a writer working in
advertising. Their relationship has become tenuous, as highlighted by the
fact that he feels the need to continuously talk and she desires silence,
particularly during sex. As a result, Jerry has decided to pursue Mary, who
is the wife of his friend Barry. Their relationship is also strained, since
her frigidity is squelching his amorousness. Meanwhile, Jerry and Barry have
another friend, Cary, who is not in a relationship. This is very
understandable since he is an emotionally sadistic jerk who is only
interested in control and domination. The final character in the story is
Cheri, an artist's assistant whom Terri meets in a gallery and develops a
sexual relationship as a response to Jerry's infidelity.

The film is an examination of the characters and the often brutal nature of
relationships. Sex is pervasive in the story, but it is not gratuitous,
because the physical intimacy experienced by partners reflects a deeper
emotional connection, and these are what the characters are missing. This is
not limited to just the couples. The three male characters have an enduring
friendship, but there are subtle homosexual overtones even here, as
witnessed by their locker room behavior.

The title of the film and the fact all the characters have similar, rhyming
names is a suggestion that these characters reflect the actual relationships
of actual people we may know. Indeed, even the close friends in this story
often do not see what transpires below the surface of the complex
relationships of their own friends and neighbors. The writing and acting are
well developed, as the characters are more than flat stereotypes. The main
weakness of the film is a lack of involvement in the story. As we are
watching the story of the characters, there is the feeling that we are
detachedly watching a laboratory experiment and waiting for its outcome.
There is no sense of involvement or empathy with the characters. Perhaps it
is because most of the characters are multi-dimensional and flawed, but this
is not a handicap for a well made film (like "Happiness").

Moderately recommended. This is a dreary story of the difficulty of
relationships. Although it has good material to work with and the characters
are well drawn, it lacks the engaging quality that most good character
dramas have.

(c) 1999 Murali Krishnan
The Art House Squatter
http://ArtHouseSquatter.com

-----------== Posted via Deja News, The Discussion Network ==----------
http://www.dejanews.com/       Search, Read, Discuss, or Start Your Own

The review above was posted to the
rec.arts.movies.reviews newsgroup (de.rec.film.kritiken for German reviews).

The Internet Movie Database accepts no responsibility for the contents of the
review and has no editorial control. Unless stated otherwise, the copyright
belongs to the author.

Please direct comments/criticisms of the review to relevant newsgroups.

Broken URLs inthe reviews are the responsibility of the author.

The formatting of the review is likely to differ from the original due
to ASCII to HTML conversion.

Related links: index of all rec.arts.movies.reviews reviews


text:
<!DOCTYPE html>

    <title>Configure wallabag</title>
    <meta charset="utf-8">
    <meta name="description" content="Documentation for wallabag" />
    <meta name="author" content="Nicolas Lœuillet">
    <link rel="icon" href="../../templates/default/themes/daux-blue/img/favicon-blue.png" type="image/x-icon">

    <meta name="apple-mobile-web-app-capable" content="yes" />
    <meta name="viewport" content="width=device-width, initial-scale=1.0">

    <link href='../../templates/default/themes/daux-blue/css/daux-blue.css' rel='stylesheet' type='text/css'>

                    wallabag

Generated by Daux.io

                        <button type="button" class="btn btn-sidebar" id="menu-spinner-button">

                        </button>

Getting started

Help wallabag

User documentation

Download and install

Save your first article

Configure wallabag

Hidden options

                                Download
GitHub Repo
Help
Documentation contributors

Configure wallabag                                                                                Fix this page

                                        Sunday, November 16, 2014

                                        9:22 PM

Configure wallabag

From the configuration menu, you can change some wallabag's options.

Some parts won't be presented here as they have a complete chapter in the documentation (RSS
feeds, import from a third party service, export your data, Epub conversion and multi-user mode).

Theme

Of tastes and colors, we do not discuss. That's why wallabag offers you to change its skin to please everyone.

Select the theme of your choice from the drop down list and confirm by clicking on Update.

Language

Select the language of your choice in the drop down list and confirm by clicking on Update
.

Password

Fill in your new password in the two dedicated fields and confirm by clicking on Update.

You will be disconnected and need to enter your new password to access your wallabag.

Clear the cache

The cache allows wallabag perform certain functions faster.

After updating wallabag, it is advised to clear the cache. You only have to click to do so.


text:
import time
import threading
import collections

import synapse.lib.sched as s_sched

from synapse.eventbus import EventBus

miss = ()
class Cache(EventBus):
    '''
    A callback driven cache with options for timeout / maxsize.

    Example:

        cache = Cache(maxtime=60)
        cache.put(iden,thing)

        # ... some time later
        valu = cache.get(iden)

    Notes:

        The maxtime option is used to specify cache time based
        flushing.  Cache accesses continue to bump a timestamp
        forward to facilite flushing the cache entries which
        were least recently requested.

    '''

    def __init__(self, maxtime=None):
        EventBus.__init__(self)

        self.sched = s_sched.getGlobSched()
        self.onmiss = None

        self.cache = {}
        self.lasthit = {}
        self.cachelock = threading.Lock()

        self.onfini( self._onCacheFini )

        self.schevt = None
        self.maxtime = maxtime

        if self.maxtime != None:
            self._checkCacheTimes()

    def setOnMiss(self, onmiss):
        '''
        Set a callback function to use on cache miss.

        Example:

            def onmiss(key):
                return stuff.get(key)

            cache.setOnMiss( onmiss )

        '''
        self.onmiss = onmiss

    def _checkCacheTimes(self):
        mintime = time.time() - self.maxtime
        try:
            hits = [ k for (k,t) in self.lasthit.items() if t < mintime ]
            [ self.pop(k) for k in hits ]
        finally:
            if not self.isfini and self.maxtime != None:
                ival = self.maxtime / 10.0
                self.sched.insec(ival, self._checkCacheTimes )

    def clear(self):
        '''
        Flush and clear the entire cache.
        '''
        [ self.flush(key) for key in self.keys() ]
        self.cache.clear()
        self.lasthit.clear()

    def get(self, key):
        '''
        Return a val from the cache.

        Example:

            val = cache.get(key)

        '''
        val = self.cache.get(key,miss)
        if val is not miss:
            self.lasthit[key] = time.time()
            return val

        if self.onmiss == None:
            return None

        with self.cachelock:
            val = self.cache.get(key,miss)
            if val is miss:
                val = self.onmiss(key)

            self.cache[key] = val
            self.lasthit[key] = time.time()
            return val

    def put(self, key, val):
        '''
        Put a key:val into the cache.

        Example:

            cache.put('woot',10)

        '''
        self.cache[key] = val
        self.lasthit[key] = time.time()
        self.fire('cache:put', key=key, val=val)

    def pop(self, key):
        '''
        Remove and return a val from the cache.

        Example:

            cache.pop('woot')

        '''
        val = self.cache.pop(key,None)

        self.lasthit.pop(key,None)

        self.fire('cache:flush', key=key, val=val)
        self.fire('cache:pop', key=key, val=val)
        return val

    def flush(self, key):
        '''
        Flush a key:val within the cache.

        Example:

            cache.flush('woot')

        Notes:

            * Mostly used to trigger "cache:flush" events
        '''
        val = self.cache.get(key)
        self.fire('cache:flush', key=key, val=val)
        return val

    def keys(self):
        '''
        Return a list of the keys in the cache.

        Example:

            for key in cache.keys():
                stuff(key)

        '''
        return list(self.cache.keys())

    def values(self):
        return list(self.cache.values())

    def __len__(self):
        return len(self.cache)

    def __iter__(self):
        return list( self.cache.items() )

    def _onCacheFini(self):
        for key in self.keys():
            self.pop(key)

        if self.schevt != None:
            self.sched.cancel(self.schevt)

class TufoCache(Cache):

    def __init__(self, core, maxtime=None):
        Cache.__init__(self, maxtime=maxtime)

        self.core = core
        self.setOnMiss( core.getTufoById )

    def _onTufoFlush(self, event):
        iden = event[1].get('key')
        tufo0 = event[1].get('val')

        tufo1 = self.core.getTufoById(iden)
        if tufo1 == None:
            return

        self.core.setTufoProps(tufo1, **tufo0[1])

class TufoPropCache(TufoCache):

    def __init__(self, core, prop, maxtime=None):
        TufoCache.__init__(self, core, maxtime=maxtime)
        self.prop = prop
        self.setOnMiss( self.getTufoByValu )

    def getTufoByValu(self, valu):
        return self.core.getTufoByProp(self.prop,valu)

def keymeth(name):
    '''
    Decorator for use with OnDem to add key callback methods.
    '''
    def keyfunc(f):
        f._keycache_name = name
        return f
    return keyfunc

class OnDem(collections.defaultdict):
    '''
    A dictionary based caching on-demand resolver.

    Example:

        class Woot(OnDem):

            @keymeth('foo')
            def _getFooThing(self):
                # only called once
                return FooThing()

        woot = Woot()
        foo = woot.get('foo')

    '''
    def __init__(self):

        collections.defaultdict.__init__(self)

        self._key_funcs = {}

        for name in dir(self):
            attr = getattr(self,name,None)
            keyn = getattr(attr,'_keycache_name',None)
            if keyn == None:
                continue

            self._key_funcs[keyn] = attr

    def __missing__(self, name):
        func = self._key_funcs.get(name)
        if func == None:
            raise KeyError(name)

        valu = func()
        self[name] = valu
        return valu

    def get(self, name):
        '''
        Return the value for the given OnDem key.

        Example:

            woot = od.get('woot')

        '''
        return self[name]

    def add(self, name, func, *args, **kwargs):
        '''
        Add a key lookup function callback to the OnDem dict.

        Example:

            def getfoo():
                return FooThing()

            od = OnDem()

            od.add('foo', getfoo)

            foo = od.get('foo')

        '''
        def keyfunc():
            return func(*args,**kwargs)

        self._key_funcs[name] = keyfunc

class KeyCache(collections.defaultdict):
    '''
    A fast key/val lookup cache.

    Example:

        cache = KeyCache( getFooThing )

        valu = cache[x]

    '''
    def __init__(self, lookmeth):
        collections.defaultdict.__init__(self)
        self.lookmeth = lookmeth

    def __missing__(self, key):
        valu = self.lookmeth(key)
        self[key] = valu
        return valu

    def pop(self, key):
        return collections.defaultdict.pop(self, key, None)

    def get(self, key):
        return self[key]

    def put(self, key, val):
        self[key] = val


text:
package jp.maju.wearsampleapplication;

import android.app.Activity;
import android.os.Bundle;
import android.support.v4.app.RemoteInput;
import android.view.Menu;
import android.view.MenuItem;
import android.widget.TextView;

public class VoiceRecieveActivity extends Activity {

    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_voice_recieve);
        Bundle result = RemoteInput.getResultsFromIntent(getIntent());
        CharSequence resultChars = result.getCharSequence("key");
        ((TextView)findViewById(R.id.text_voice)).setText(resultChars);
    }

    @Override
    public boolean onCreateOptionsMenu(Menu menu) {
        // Inflate the menu; this adds items to the action bar if it is present.
        getMenuInflater().inflate(R.menu.voice_recieve, menu);
        return true;
    }

    @Override
    public boolean onOptionsItemSelected(MenuItem item) {
        // Handle action bar item clicks here. The action bar will
        // automatically handle clicks on the Home/Up button, so long
        // as you specify a parent activity in AndroidManifest.xml.
        int id = item.getItemId();
        if (id == R.id.action_settings) {
            return true;
        }
        return super.onOptionsItemSelected(item);
    }
}


text:
var nano        = require('nano');
var cookie      = require('cookie');
var config      = require('../../config/couchdb');
var publicCouch = nano(config.url);

module.exports = privileged;

var privilegedCouch;
var loggingIn = false;
var queue = [];

function privileged(dbName, cb) {
  if (! cb && 'function' == typeof dbName) {
    cb = dbName;
    dbName = undefined;
  }

  if (privilegedCouch) {
    var db = dbName ? privilegedCouch.use(dbName) : privilegedCouch;
    cb(null, db);
  } else {
    login(function(err, db) {
      if (err) {
        console.error(err.stack || err);
        return cb(err);
      };
      db = dbName ? db.use(dbName) : db;
      cb(err, db);
    });
  }
}

function login(cb) {
  queue.push(cb);

  if (! loggingIn) {
    loggingIn = true;
    publicCouch.auth(config.admin.username, config.admin.password, replied);
  }
}

function replied(err, res, headers) {
  console.log('privileged replied: %j'.red, arguments);
  var sessionId, header;

  loggingIn = false;

  if (err) globalError(err);
  else {

    // HACK: use basic auth in URL all the time
    // because I think cookies are not working in nano
    privilegedCouch = nano(config.admin_url);
    replyAll(null, privilegedCouch);
    return;

    header = headers['set-cookie'][0];
    if (header) sessionId = cookie.parse(header).AuthSession;
    console.log('COUICHDB AUTH SESSION:', sessionId);
    if (! sessionId) globalError(new Error('no AuthSession cookie found'));
    else {
      privilegedCouch = nano({
        url: config.url,
        cookie: 'AuthSession=' + encodeURIComponent(sessionId)
      });
      replyAll(null, privilegedCouch);
    }
  }
}

function globalError(err) {
  replyAll(err);
}

function replyAll() {
  while(queue.length) queue.shift().apply(null, arguments);
}


text:
package org.simpleflatmapper.util.test;

import org.junit.Test;
import org.simpleflatmapper.util.NullConsumer;

import static org.junit.Assert.*;

public class NullConsumerTest {

    @Test
    public void testNullConsumer() {
        NullConsumer.INSTANCE.accept(null);
    }
}


text:
using System;
using System.ComponentModel.Composition;
using System.Diagnostics;
using System.Globalization;
using System.Reactive.Linq;
using GitHub.Authentication;
using GitHub.Exports;
using GitHub.Info;
using GitHub.Services;
using GitHub.Validation;
using NullGuard;
using Octokit;
using ReactiveUI;

namespace GitHub.ViewModels
{
    [ExportViewModel(ViewType = UIViewType.TwoFactor)]
    [PartCreationPolicy(CreationPolicy.NonShared)]
    public class TwoFactorDialogViewModel : BaseViewModel, ITwoFactorDialogViewModel
    {
        bool isAuthenticationCodeSent;
        bool isBusy;
        bool invalidAuthenticationCode;
        string authenticationCode;
        TwoFactorType twoFactorType;
        readonly ObservableAsPropertyHelper<string> description;
        readonly ObservableAsPropertyHelper<bool> isSms;
        readonly ObservableAsPropertyHelper<bool> showErrorMessage;

        [ImportingConstructor]
        public TwoFactorDialogViewModel(
            IVisualStudioBrowser browser,
            ITwoFactorChallengeHandler twoFactorChallengeHandler)
        {
            Title = Resources.TwoFactorTitle;
            twoFactorChallengeHandler.SetViewModel(this);

            var canVerify = this.WhenAny(
                x => x.AuthenticationCode,
                x => x.IsBusy,
                (code, busy) => !string.IsNullOrEmpty(code.Value) && code.Value.Length == 6 && !busy.Value);

            OkCommand = ReactiveCommand.Create(canVerify);
            CancelCommand = ReactiveCommand.Create();
            NavigateLearnMore = ReactiveCommand.Create();
            NavigateLearnMore.Subscribe(x => browser.OpenUrl(GitHubUrls.TwoFactorLearnMore));
            //TODO: ShowHelpCommand.Subscribe(x => browser.OpenUrl(twoFactorHelpUri));
            ResendCodeCommand = ReactiveCommand.Create();

            showErrorMessage = this.WhenAny(
                x => x.IsAuthenticationCodeSent,
                x => x.InvalidAuthenticationCode,
                (authSent, invalid) => invalid.Value && !authSent.Value)
                .ToProperty(this, x => x.ShowErrorMessage);

            description = this.WhenAny(x => x.TwoFactorType, x => x.Value)
                .Select(type =>
                {
                    switch (type)
                    {
                        case TwoFactorType.Sms:
                            return Resources.TwoFactorSms;
                        case TwoFactorType.AuthenticatorApp:
                            return Resources.TwoFactorApp;
                        case TwoFactorType.Unknown:
                            return Resources.TwoFactorUnknown;

                        default:
                            return null;
                    }
                })
                .ToProperty(this, x => x.Description);

            isShowing = this.WhenAny(x => x.TwoFactorType, x => x.Value)
                .Select(factorType => factorType != TwoFactorType.None)
                .ToProperty(this, x => x.IsShowing);

            isSms = this.WhenAny(x => x.TwoFactorType, x => x.Value)
                .Select(factorType => factorType == TwoFactorType.Sms)
                .ToProperty(this, x => x.IsSms);
        }

        public IObservable<RecoveryOptionResult> Show(UserError userError)
        {
            IsBusy = false;
            var error = userError as TwoFactorRequiredUserError;
            Debug.Assert(error != null,
                String.Format(CultureInfo.InvariantCulture, "The user error is '{0}' not a TwoFactorRequiredUserError", userError));
            InvalidAuthenticationCode = error.RetryFailed;
            TwoFactorType = error.TwoFactorType;
            var ok = OkCommand
                .Do(_ => IsBusy = true)
                .Select(_ => AuthenticationCode == null
                    ? RecoveryOptionResult.CancelOperation
                    : RecoveryOptionResult.RetryOperation)
                .Do(_ => error.ChallengeResult = AuthenticationCode != null
                    ? new TwoFactorChallengeResult(AuthenticationCode)
                    : null);
            var resend = ResendCodeCommand.Select(_ => RecoveryOptionResult.RetryOperation)
                .Do(_ => error.ChallengeResult = TwoFactorChallengeResult.RequestResendCode);
            var cancel = CancelCommand.Select(_ => RecoveryOptionResult.CancelOperation);
            return Observable.Merge(ok, cancel, resend)
                .Take(1)
                .Do(_ => IsAuthenticationCodeSent = error.ChallengeResult == TwoFactorChallengeResult.RequestResendCode);
        }

        public TwoFactorType TwoFactorType
        {
            get { return twoFactorType; }
            private set { this.RaiseAndSetIfChanged(ref twoFactorType, value); }
        }

        public bool IsSms { get { return isSms.Value; } }

        public bool IsAuthenticationCodeSent
        {
            get { return isAuthenticationCodeSent; }
            private set { this.RaiseAndSetIfChanged(ref isAuthenticationCodeSent, value); }
        }

        public string Description
        {
            [return: AllowNull]
            get { return description.Value; }
        }

        [AllowNull]
        public string AuthenticationCode
        {
            [return: AllowNull]
            get { return authenticationCode; }
            set { this.RaiseAndSetIfChanged(ref authenticationCode, value); }
        }

        public ReactiveCommand<object> OkCommand { get; private set; }
        public ReactiveCommand<object> NavigateLearnMore { get; private set; }
        public ReactiveCommand<object> ResendCodeCommand { get; private set; }
        public ReactivePropertyValidator AuthenticationCodeValidator { get; private set; }

        public bool InvalidAuthenticationCode
        {
            get { return invalidAuthenticationCode; }
            private set { this.RaiseAndSetIfChanged(ref invalidAuthenticationCode, value); }
        }

        public bool ShowErrorMessage
        {
            get { return showErrorMessage.Value; }
        }

        public bool IsBusy
        {
            get { return isBusy; }
            set { this.RaiseAndSetIfChanged(ref isBusy, value); }
        }
    }
}


text:
void shuffle(int *a, size_t len) {
    for (size_t i = 0; i < len - 2; i++) {
        size_t j = rand() % (len - i) + i;
        int temp = a[i];
        a[i] = a[j];
        a[j] = temp;
    }
}

void rs_correct_encode(void *encoder, uint8_t *msg, size_t msg_length,
                       uint8_t *msg_out) {
    correct_reed_solomon_encode((correct_reed_solomon *)encoder, msg,
                                msg_length, msg_out);
}

void rs_correct_decode(void *decoder, uint8_t *encoded, size_t encoded_length,
                       uint8_t *erasure_locations, size_t erasure_length,
                       uint8_t *msg, size_t pad_length, size_t num_roots) {
    correct_reed_solomon_decode_with_erasures(
        (correct_reed_solomon *)decoder, encoded, encoded_length,
        erasure_locations, erasure_length, msg);
}

rs_testbench *rs_testbench_create(size_t block_length, size_t min_distance) {
    rs_testbench *testbench = calloc(1, sizeof(rs_testbench));

    size_t message_length = block_length - min_distance;
    testbench->message_length = message_length;
    testbench->block_length = block_length;
    testbench->min_distance = min_distance;

    testbench->msg = calloc(message_length, sizeof(unsigned char));
    testbench->encoded = malloc(block_length * sizeof(uint8_t));

    testbench->indices = malloc(block_length * sizeof(int));

    testbench->corrupted_encoded = malloc(block_length * sizeof(uint8_t));
    testbench->erasure_locations = malloc(min_distance * sizeof(uint8_t));
    testbench->recvmsg = malloc(sizeof(unsigned char) * message_length);

    return testbench;
}

void rs_testbench_destroy(rs_testbench *testbench) {
    free(testbench->msg);
    free(testbench->encoded);
    free(testbench->indices);
    free(testbench->corrupted_encoded);
    free(testbench->erasure_locations);
    free(testbench->recvmsg);
    free(testbench);
}

rs_test_run test_rs_errors(rs_test *test, rs_testbench *testbench, size_t msg_length,
                    size_t num_errors, size_t num_erasures) {
    rs_test_run run;
    run.output_matches = false;

    if (msg_length > testbench->message_length) {
        return run;
    }

    for (size_t i = 0; i < msg_length; i++) {
        testbench->msg[i] = rand() % 256;
    }

    size_t block_length = msg_length + testbench->min_distance;
    size_t pad_length = testbench->message_length - msg_length;

    test->encode(test->encoder, testbench->msg, msg_length, testbench->encoded);

    memcpy(testbench->corrupted_encoded, testbench->encoded, block_length);

    for (int i = 0; i < block_length; i++) {
        testbench->indices[i] = i;
    }

    shuffle(testbench->indices, block_length);

    for (unsigned int i = 0; i < num_erasures; i++) {
        int index = testbench->indices[i];
        uint8_t corruption_mask = (rand() % 255) + 1;
        testbench->corrupted_encoded[index] ^= corruption_mask;
        testbench->erasure_locations[i] = index;
    }

    for (unsigned int i = 0; i < num_errors; i++) {
        int index = testbench->indices[i + num_erasures];
        uint8_t corruption_mask = (rand() % 255) + 1;
        testbench->corrupted_encoded[index] ^= corruption_mask;
    }

    test->decode(test->decoder, testbench->corrupted_encoded, block_length,
                 testbench->erasure_locations, num_erasures,
                 testbench->recvmsg, pad_length, testbench->min_distance);

    run.output_matches = (bool)(memcmp(testbench->msg, testbench->recvmsg, msg_length) == 0);

    return run;
}


text:
package org.apache.ambari.server.actionmanager;

import java.lang.reflect.Type;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.TreeMap;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

import javax.persistence.EntityManager;

import org.apache.ambari.server.AmbariException;
import org.apache.ambari.server.ClusterNotFoundException;
import org.apache.ambari.server.Role;
import org.apache.ambari.server.RoleCommand;
import org.apache.ambari.server.ServiceComponentHostNotFoundException;
import org.apache.ambari.server.ServiceComponentNotFoundException;
import org.apache.ambari.server.agent.AgentCommand;
import org.apache.ambari.server.agent.CancelCommand;
import org.apache.ambari.server.agent.CommandReport;
import org.apache.ambari.server.agent.ExecutionCommand;
import org.apache.ambari.server.configuration.Configuration;
import org.apache.ambari.server.controller.HostsMap;
import org.apache.ambari.server.events.ActionFinalReportReceivedEvent;
import org.apache.ambari.server.events.jpa.EntityManagerCacheInvalidationEvent;
import org.apache.ambari.server.events.listeners.tasks.TaskStatusListener;
import org.apache.ambari.server.events.publishers.AgentCommandsPublisher;
import org.apache.ambari.server.events.publishers.AmbariEventPublisher;
import org.apache.ambari.server.events.publishers.JPAEventPublisher;
import org.apache.ambari.server.metadata.RoleCommandOrder;
import org.apache.ambari.server.metadata.RoleCommandOrderProvider;
import org.apache.ambari.server.metadata.RoleCommandPair;
import org.apache.ambari.server.orm.dao.HostRoleCommandDAO;
import org.apache.ambari.server.orm.entities.HostRoleCommandEntity;
import org.apache.ambari.server.orm.entities.RequestEntity;
import org.apache.ambari.server.serveraction.ServerActionExecutor;
import org.apache.ambari.server.state.Cluster;
import org.apache.ambari.server.state.Clusters;
import org.apache.ambari.server.state.Host;
import org.apache.ambari.server.state.HostState;
import org.apache.ambari.server.state.Service;
import org.apache.ambari.server.state.ServiceComponent;
import org.apache.ambari.server.state.ServiceComponentHost;
import org.apache.ambari.server.state.ServiceComponentHostEvent;
import org.apache.ambari.server.state.fsm.InvalidStateTransitionException;
import org.apache.ambari.server.state.svccomphost.ServiceComponentHostOpFailedEvent;
import org.apache.ambari.server.utils.StageUtils;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.base.Function;
import com.google.common.cache.Cache;
import com.google.common.cache.CacheBuilder;
import com.google.common.collect.ArrayListMultimap;
import com.google.common.collect.ImmutableSet;
import com.google.common.collect.ListMultimap;
import com.google.common.collect.Lists;
import com.google.common.collect.Multimap;
import com.google.common.eventbus.Subscribe;
import com.google.common.reflect.TypeToken;
import com.google.inject.Inject;
import com.google.inject.Provider;
import com.google.inject.Singleton;
import com.google.inject.name.Named;
import com.google.inject.persist.UnitOfWork;

/**
 * This class encapsulates the action scheduler thread.
 * Action schedule frequently looks at action database and determines if
 * there is an action that can be scheduled.
 */
@Singleton
class ActionScheduler implements Runnable {

  private static final Logger LOG = LoggerFactory.getLogger(ActionScheduler.class);

  public static final String FAILED_TASK_ABORT_REASONING =
    "Server considered task failed and automatically aborted it";

  @Inject
  private RoleCommandOrderProvider roleCommandOrderProvider;

  @Inject
  private UnitOfWork unitOfWork;

  @Inject
  private Clusters clusters;

  @Inject
  private AmbariEventPublisher ambariEventPublisher;

  @Inject
  private HostsMap hostsMap;

  @Inject
  private Configuration configuration;

  @Inject
  Provider<EntityManager> entityManagerProvider;

  /**
   * Used for turning instances of {@link HostRoleCommandEntity} into
   * {@link HostRoleCommand}.
   */
  @Inject
  private HostRoleCommandFactory hostRoleCommandFactory;

  /**
   * Used for retrieving {@link HostRoleCommandEntity} instances which need to
   * be cancelled.
   */
  @Inject
  private HostRoleCommandDAO hostRoleCommandDAO;

  @Inject
  private AgentCommandsPublisher agentCommandsPublisher;

  /**
   * The current thread's reference to the {@link EntityManager}.
   */
  volatile EntityManager threadEntityManager;

  private final long actionTimeout;
  private final long sleepTime;
  private volatile boolean shouldRun = true;
  private Thread schedulerThread = null;
  private final ActionDBAccessor db;
  private short maxAttempts = 2;
  private final JPAEventPublisher jpaPublisher;
  private boolean taskTimeoutAdjustment = true;
  private final Object wakeupSyncObject = new Object();
  private final ServerActionExecutor serverActionExecutor;

  private final Set<Long> requestsInProgress = new HashSet<>();

  /**
   * Contains request ids that have been scheduled to be cancelled,
   * but are not cancelled yet
   */
  private final Set<Long> requestsToBeCancelled =
    Collections.newSetFromMap(new ConcurrentHashMap<Long, Boolean>());

  /**
   * Maps request IDs to reasoning for cancelling request.
   * Map is NOT synchronized, so any access to it should synchronize on
   * requestsToBeCancelled object
   */
  private final Map<Long, String> requestCancelReasons =
    new HashMap<>();

  /**
   * true if scheduler should run ASAP.
   * We need this flag to avoid sleep in situations, when
   * we receive awake() request during running a scheduler iteration.
   */
  private boolean activeAwakeRequest = false;

  private AtomicBoolean taskStatusLoaded = new AtomicBoolean();

  //Cache for clusterHostinfo, key - stageId-requestId
  private Cache<String, Map<String, Set<String>>> clusterHostInfoCache;
  private Cache<String, Map<String, String>> commandParamsStageCache;
  private Cache<String, Map<String, String>> hostParamsStageCache;

  /**
   * Guice-injected Constructor.
   *
   * @param sleepTime
   * @param actionTimeout
   * @param db
   * @param jpaPublisher
   */
  @Inject
  public ActionScheduler(@Named("schedulerSleeptime") long sleepTime,
                         @Named("actionTimeout") long actionTimeout, ActionDBAccessor db,
                         JPAEventPublisher jpaPublisher) {

    this.sleepTime = sleepTime;
    this.actionTimeout = actionTimeout;
    this.db = db;

    this.jpaPublisher = jpaPublisher;
    this.jpaPublisher.register(this);

    serverActionExecutor = new ServerActionExecutor(db, sleepTime);

    initializeCaches();
  }

  /**
   * Unit Test Constructor.
   * @param sleepTimeMilliSec
   * @param actionTimeoutMilliSec
   * @param db
   * @param fsmObject
   * @param maxAttempts
   * @param hostsMap
   * @param unitOfWork
   * @param ambariEventPublisher
   * @param configuration
   * @param entityManagerProvider
   * @param hostRoleCommandDAO
   * @param hostRoleCommandFactory
   * @param roleCommandOrderProvider
   */
  protected ActionScheduler(long sleepTimeMilliSec, long actionTimeoutMilliSec, ActionDBAccessor db,
                            Clusters fsmObject, int maxAttempts, HostsMap hostsMap,
                            UnitOfWork unitOfWork, AmbariEventPublisher ambariEventPublisher,
                            Configuration configuration, Provider<EntityManager> entityManagerProvider,
                            HostRoleCommandDAO hostRoleCommandDAO, HostRoleCommandFactory hostRoleCommandFactory,
                            RoleCommandOrderProvider roleCommandOrderProvider, AgentCommandsPublisher agentCommandsPublisher) {

    sleepTime = sleepTimeMilliSec;
    actionTimeout = actionTimeoutMilliSec;
    this.db = db;
    clusters = fsmObject;
    this.maxAttempts = (short) maxAttempts;
    this.hostsMap = hostsMap;
    this.unitOfWork = unitOfWork;
    this.ambariEventPublisher = ambariEventPublisher;
    this.configuration = configuration;
    this.entityManagerProvider = entityManagerProvider;
    this.hostRoleCommandDAO = hostRoleCommandDAO;
    this.hostRoleCommandFactory = hostRoleCommandFactory;
    jpaPublisher = null;
    this.roleCommandOrderProvider = roleCommandOrderProvider;
    this.agentCommandsPublisher = agentCommandsPublisher;

    serverActionExecutor = new ServerActionExecutor(db, sleepTime);
    initializeCaches();
  }

  /**
   * Unit Test Constructor.
   *
   * @param sleepTimeMilliSec
   * @param actionTimeoutMilliSec
   * @param db
   * @param fsmObject
   * @param maxAttempts
   * @param hostsMap
   * @param unitOfWork
   * @param ambariEventPublisher
   * @param configuration
   * @param hostRoleCommandDAO
   * @param hostRoleCommandFactory
   */
  protected ActionScheduler(long sleepTimeMilliSec, long actionTimeoutMilliSec, ActionDBAccessor db,
                            Clusters fsmObject, int maxAttempts, HostsMap hostsMap,
                            UnitOfWork unitOfWork, AmbariEventPublisher ambariEventPublisher,
                            Configuration configuration, Provider<EntityManager> entityManagerProvider,
                            HostRoleCommandDAO hostRoleCommandDAO, HostRoleCommandFactory hostRoleCommandFactory,
                            AgentCommandsPublisher agentCommandsPublisher) {

    this(sleepTimeMilliSec, actionTimeoutMilliSec, db, fsmObject, maxAttempts, hostsMap, unitOfWork,
            ambariEventPublisher, configuration, entityManagerProvider, hostRoleCommandDAO, hostRoleCommandFactory,
            null, agentCommandsPublisher);
  }

  /**
   * Initializes the caches.
   */
  private void initializeCaches() {
    clusterHostInfoCache = CacheBuilder.newBuilder().
      expireAfterAccess(5, TimeUnit.MINUTES).
      build();

    commandParamsStageCache = CacheBuilder.newBuilder().
      expireAfterAccess(5, TimeUnit.MINUTES).
      build();

    hostParamsStageCache = CacheBuilder.newBuilder().
      expireAfterAccess(5, TimeUnit.MINUTES).
      build();
  }

  public void start() {
    schedulerThread = new Thread(this, "ambari-action-scheduler");
    schedulerThread.start();

    // Start up the ServerActionExecutor. Since it is directly related to the ActionScheduler it
    // should be started and stopped along with it.
    serverActionExecutor.start();
  }

  public void stop() {
    shouldRun = false;
    schedulerThread.interrupt();

    // Stop the ServerActionExecutor. Since it is directly related to the ActionScheduler it should
    // be started and stopped along with it.
    serverActionExecutor.stop();
  }

  /**
   * Should be called from another thread when we want scheduler to
   * make a run ASAP (for example, to process desired configs of SCHs).
   * The method is guaranteed to return quickly.
   */
  public void awake() {
    synchronized (wakeupSyncObject) {
      activeAwakeRequest = true;
      wakeupSyncObject.notify();
    }
  }

  @Override
  public void run() {
    while (shouldRun) {
      try {
        synchronized (wakeupSyncObject) {
          if (!activeAwakeRequest) {
            wakeupSyncObject.wait(sleepTime);
          }
          activeAwakeRequest = false;
        }

        doWork();

      } catch (InterruptedException ex) {
        LOG.warn("Scheduler thread is interrupted going to stop", ex);
        shouldRun = false;
      } catch (Exception ex) {
        LOG.warn("Exception received", ex);
        requestsInProgress.clear();
      } catch (Throwable t) {
        LOG.warn("ERROR", t);
        requestsInProgress.clear();
      }
    }
  }

  public void doWork() throws AmbariException {
    try {
      unitOfWork.begin();

      // grab a reference to this UnitOfWork's EM
      threadEntityManager = entityManagerProvider.get();

      // The first thing to do is to abort requests that are cancelled
      processCancelledRequestsList();

      // !!! getting the stages in progress could be a very expensive call due
      // to the join being used; there's no need to make it if there are
      // no commands in progress
      if (db.getCommandsInProgressCount() == 0) {
        // Nothing to do
        if (LOG.isDebugEnabled()) {
          LOG.debug("There are no stages currently in progress.");
        }

        return;
      }

      Set<Long> runningRequestIds = new HashSet<>();
      List<Stage> firstStageInProgressPerRequest = db.getFirstStageInProgressPerRequest();
      if (LOG.isDebugEnabled()) {
        LOG.debug("Scheduler wakes up");
        LOG.debug("Processing {} in progress stages", firstStageInProgressPerRequest.size());
      }

      publishInProgressTasks(firstStageInProgressPerRequest);

      if (firstStageInProgressPerRequest.isEmpty()) {
        // Nothing to do
        if (LOG.isDebugEnabled()) {
          LOG.debug("There are no stages currently in progress.");
        }

        return;
      }

      int i_stage = 0;

      // filter the stages in progress down to those which can be scheduled in
      // parallel
      List<Stage> stages = filterParallelPerHostStages(firstStageInProgressPerRequest);

      boolean exclusiveRequestIsGoing = false;
      // This loop greatly depends on the fact that order of stages in
      // a list does not change between invocations
      for (Stage stage : stages) {
        // Check if we can process this stage in parallel with another stages
        i_stage++;
        long requestId = stage.getRequestId();
        LOG.debug("==> STAGE_i = {}(requestId={},StageId={})", i_stage, requestId, stage.getStageId());

        RequestEntity request = db.getRequestEntity(requestId);

        if (request.isExclusive()) {
          if (runningRequestIds.size() > 0) {
            // As a result, we will wait until any previous stages are finished
            LOG.debug("Stage requires exclusive execution, but other requests are already executing. Stopping for now");
            break;
          }
          exclusiveRequestIsGoing = true;
        }

        if (runningRequestIds.contains(requestId)) {
          // We don't want to process different stages from the same request in parallel
          LOG.debug("==> We don't want to process different stages from the same request in parallel");
          continue;
        } else {
          runningRequestIds.add(requestId);
          if (!requestsInProgress.contains(requestId)) {
            requestsInProgress.add(requestId);
            db.startRequest(requestId);
          }
        }

        // Commands that will be scheduled in current scheduler wakeup
        List<ExecutionCommand> commandsToSchedule = new ArrayList<>();
        Multimap<Long, AgentCommand> commandsToEnqueue = ArrayListMultimap.create();

        Map<String, RoleStats> roleStats =
          processInProgressStage(stage, commandsToSchedule, commandsToEnqueue);

        // Check if stage is failed
        boolean failed = false;
        for (Map.Entry<String, RoleStats> entry : roleStats.entrySet()) {

          String role = entry.getKey();
          RoleStats stats = entry.getValue();

          if (LOG.isDebugEnabled()) {
            LOG.debug("Stats for role: {}, stats={}", role, stats);
          }

          // only fail the request if the role failed and the stage is not
          // skippable
          if (stats.isRoleFailed() && !stage.isSkippable()) {
            LOG.warn("{} failed, request {} will be aborted", role, request.getRequestId());

            failed = true;
            break;
          }
        }

        if (!failed) {
          // Prior stage may have failed and it may need to fail the whole request
          failed = hasPreviousStageFailed(stage);
        }

        if (failed) {
          LOG.error("Operation completely failed, aborting request id: {}", stage.getRequestId());
          cancelHostRoleCommands(stage.getOrderedHostRoleCommands(), FAILED_TASK_ABORT_REASONING);
          abortOperationsForStage(stage);
          return;
        }

        List<ExecutionCommand> commandsToStart = new ArrayList<>();
        List<ExecutionCommand> commandsToUpdate = new ArrayList<>();

        //Schedule what we have so far

        for (ExecutionCommand cmd : commandsToSchedule) {
          processHostRole(request, stage, cmd, commandsToStart, commandsToUpdate);
        }

        LOG.debug("==> Commands to start: {}", commandsToStart.size());
        LOG.debug("==> Commands to update: {}", commandsToUpdate.size());

        //Multimap is analog of Map<Object, List<Object>> but allows to avoid nested loop
        ListMultimap<String, ServiceComponentHostEvent> eventMap = formEventMap(stage, commandsToStart);
        Map<ExecutionCommand, String> commandsToAbort = new HashMap<>();
        if (!eventMap.isEmpty()) {
          LOG.debug("==> processing {} serviceComponentHostEvents...", eventMap.size());
          Cluster cluster = clusters.getCluster(stage.getClusterName());
          if (cluster != null) {
            Map<ServiceComponentHostEvent, String> failedEvents = cluster.processServiceComponentHostEvents(eventMap);

            if (failedEvents.size() > 0) {
              LOG.error("==> {} events failed.", failedEvents.size());
            }

            for (Iterator<ExecutionCommand> iterator = commandsToUpdate.iterator(); iterator.hasNext(); ) {
              ExecutionCommand cmd = iterator.next();
              for (ServiceComponentHostEvent event : failedEvents.keySet()) {
                if (StringUtils.equals(event.getHostName(), cmd.getHostname()) &&
                  StringUtils.equals(event.getServiceComponentName(), cmd.getRole())) {
                  iterator.remove();
                  commandsToAbort.put(cmd, failedEvents.get(event));
                  break;
                }
              }
            }
          } else {
            LOG.warn("There was events to process but cluster {} not found", stage.getClusterName());
          }
        }

        LOG.debug("==> Scheduling {} tasks...", commandsToUpdate.size());
        db.bulkHostRoleScheduled(stage, commandsToUpdate);

        if (commandsToAbort.size() > 0) { // Code branch may be a bit slow, but is extremely rarely used
          LOG.debug("==> Aborting {} tasks...", commandsToAbort.size());
          // Build a list of HostRoleCommands
          List<Long> taskIds = new ArrayList<>();
          for (ExecutionCommand command : commandsToAbort.keySet()) {
            taskIds.add(command.getTaskId());
          }
          Collection<HostRoleCommand> hostRoleCommands = db.getTasks(taskIds);

          cancelHostRoleCommands(hostRoleCommands, FAILED_TASK_ABORT_REASONING);
          db.bulkAbortHostRole(stage, commandsToAbort);
        }

        LOG.debug("==> Adding {} tasks to queue...", commandsToUpdate.size());
        for (ExecutionCommand cmd : commandsToUpdate) {
          // Do not queue up server actions; however if we encounter one, wake up the ServerActionExecutor
          if (Role.AMBARI_SERVER_ACTION.name().equals(cmd.getRole())) {
            serverActionExecutor.awake();
          } else {
            commandsToEnqueue.put(clusters.getHost(cmd.getHostname()).getHostId(), cmd);
          }
        }
        if (!commandsToEnqueue.isEmpty()) {
          agentCommandsPublisher.sendAgentCommand(commandsToEnqueue);
        }
        LOG.debug("==> Finished.");

        if (!configuration.getParallelStageExecution()) { // If disabled
          return;
        }

        if (exclusiveRequestIsGoing) {
          // As a result, we will prevent any further stages from being executed
          LOG.debug("Stage requires exclusive execution, skipping all executing any further stages");
          break;
        }
      }

      requestsInProgress.retainAll(runningRequestIds);

    } finally {
      LOG.debug("Scheduler finished work.");
      unitOfWork.end();
    }
  }

  /**
   * publish event to load {@link TaskStatusListener#activeTasksMap} {@link TaskStatusListener#activeStageMap}
   * and {@link TaskStatusListener#activeRequestMap} for all running request once during server startup.
   * This is required as some tasks may have been in progress when server was last stopped
   * @param stages list of stages
   */
  private void publishInProgressTasks(List<Stage> stages) {
    if (taskStatusLoaded.compareAndSet(false, true)) {
      if (!stages.isEmpty()) {
        Function<Stage, Long> transform = new Function<Stage, Long>() {
          @Override
          public Long apply(Stage stage) {
            return stage.getRequestId();
          }
        };
        Set<Long> runningRequestID = ImmutableSet.copyOf(Lists.transform(stages, transform));
        List<HostRoleCommand> hostRoleCommands = db.getAllTasksByRequestIds(runningRequestID);
        hostRoleCommandDAO.publishTaskCreateEvent(hostRoleCommands);
      }
    }
  }

  /**
   * Returns filtered list of stages such that the returned list is an ordered
   * list of stages that may be executed in parallel or in the order in which
   * they are presented.
   * <p/>
   * The specified stages must be ordered by request ID and may only contain the
   * next stage in progress per request (as returned by
   * {@link ActionDBAccessor#getFirstStageInProgressPerRequest()}. This is
   * because there is a requirement that within a request, no two stages may
   * ever run in parallel.
   * <p/>
   * The following rules will be applied to the list:
   * <ul>
   * <li>Stages are filtered such that the first stage in the list (assumed to
   * be the first pending stage from the earliest active request) has priority.
   * </li>
   * <li>No stage in any request may be executed before an earlier stage in the
   * same request. This requirement is automatically covered by virtue of the
   * supplied stages only being for the next stage in progress per request.</li>
   * <li>A stage in different request may be performed in parallel
   * if-and-only-if the relevant hosts for the stage in the later requests do
   * not intersect with the union of hosts from (pending) stages in earlier
   * requests. In order to accomplish this</li>
   * </ul>
   *
   * @param firstStageInProgressPerRequest
   *          the stages to process, one stage per request
   * @return a list of stages that may be executed in parallel
   */
  private List<Stage> filterParallelPerHostStages(List<Stage> firstStageInProgressPerRequest) {
    // if there's only 1 stage in progress in 1 request, simply return that stage
    if (firstStageInProgressPerRequest.size() == 1) {
      return firstStageInProgressPerRequest;
    }

    List<Stage> retVal = new ArrayList<>();

    // set the lower range (inclusive) of requests to limit the query a bit
    // since there can be a LOT of commands
    long lowerRequestIdInclusive = firstStageInProgressPerRequest.get(0).getRequestId();

    // determine if this stage can be scheduled in parallel with the other
    // stages from other requests
    for (Stage stage : firstStageInProgressPerRequest) {
      long requestId = stage.getRequestId();

      if (LOG.isTraceEnabled()) {
        LOG.trace("==> Processing stage: {}/{} ({}) for {}", requestId, stage.getStageId(), stage.getRequestContext());
      }

      boolean addStage = true;

      // there are at least 2 request in progress concurrently; determine which
      // hosts are affected
      HashSet<String> hostsInProgressForEarlierRequests = new HashSet<>(
          hostRoleCommandDAO.getBlockingHostsForRequest(lowerRequestIdInclusive, requestId));

      // Iterate over the relevant hosts for this stage to see if any intersect with the set of
      // hosts needed for previous stages.  If any intersection occurs, this stage may not be
      // executed in parallel.
      for (String host : stage.getHosts()) {
        LOG.trace("===> Processing Host {}", host);

        if (hostsInProgressForEarlierRequests.contains(host)) {
          if (LOG.isTraceEnabled()) {
            LOG.trace("===>  Skipping stage since it utilizes at least one host that a previous stage requires: {}/{} ({})", stage.getRequestId(), stage.getStageId(), stage.getRequestContext());
          }

          addStage = false;
          break;
        }
      }

      // add the stage is no other prior stages for prior requests intersect the
      // hosts in this stage
      if (addStage) {
        if (LOG.isTraceEnabled()) {
          LOG.trace("===>  Adding stage to return value: {}/{} ({})", stage.getRequestId(), stage.getStageId(), stage.getRequestContext());
        }

        retVal.add(stage);
      }
    }

    return retVal;
  }

  private boolean hasPreviousStageFailed(Stage stage) {
    boolean failed = false;

    long prevStageId = stage.getStageId() - 1;

    if (prevStageId >= 0) {
      // Find previous stage instance
      String actionId = StageUtils.getActionId(stage.getRequestId(), prevStageId);
      Stage prevStage = db.getStage(actionId);

      // If the previous stage is skippable then we shouldn't automatically fail the given stage
      if (prevStage == null || prevStage.isSkippable()) {
        return false;
      }

      Map<Role, Integer> hostCountsForRoles = new HashMap<>();
      Map<Role, Integer> failedHostCountsForRoles = new HashMap<>();

      for (String host : prevStage.getHostRoleCommands().keySet()) {
        Map<String, HostRoleCommand> roleCommandMap = prevStage.getHostRoleCommands().get(host);
        for (String role : roleCommandMap.keySet()) {
          HostRoleCommand c = roleCommandMap.get(role);
          if (hostCountsForRoles.get(c.getRole()) == null) {
            hostCountsForRoles.put(c.getRole(), 0);
            failedHostCountsForRoles.put(c.getRole(), 0);
          }
          int hostCount = hostCountsForRoles.get(c.getRole());
          hostCountsForRoles.put(c.getRole(), hostCount + 1);
          if (c.getStatus().isFailedAndNotSkippableState()) {
            int failedHostCount = failedHostCountsForRoles.get(c.getRole());
            failedHostCountsForRoles.put(c.getRole(), failedHostCount + 1);
          }
        }
      }

      for (Role role : hostCountsForRoles.keySet()) {
        float failedHosts = failedHostCountsForRoles.get(role);
        float totalHosts = hostCountsForRoles.get(role);
        if (((totalHosts - failedHosts) / totalHosts) < prevStage.getSuccessFactor(role)) {
          failed = true;
        }
      }
    }
    return failed;
  }

  /**
   * This method processes command timeouts and retry attempts, and
   * adds new (pending) execution commands to commandsToSchedule list.
   * @return the stats for the roles in the stage which are used to determine
   * whether stage has succeeded or failed
   */
  protected Map<String, RoleStats> processInProgressStage(Stage s, List<ExecutionCommand> commandsToSchedule,
                                                          Multimap<Long, AgentCommand> commandsToEnqueue) throws AmbariException {
    LOG.debug("==> Collecting commands to schedule...");
    // Map to track role status
    Map<String, RoleStats> roleStats = initRoleStats(s);
    long now = System.currentTimeMillis();
    Set<RoleCommandPair> rolesCommandsInProgress = s.getHostRolesInProgress();

    Cluster cluster = null;
    if (null != s.getClusterName()) {
      cluster = clusters.getCluster(s.getClusterName());
    }

    for (String host : s.getHosts()) {

      List<ExecutionCommandWrapper> commandWrappers = s.getExecutionCommands(host);
      Host hostObj = null;
      try {
        hostObj = clusters.getHost(host);
      } catch (AmbariException e) {
        LOG.debug("Host {} not found, stage is likely a server side action", host);
      }

      int i_my = 0;
      LOG.trace("===>host={}", host);

      for (ExecutionCommandWrapper wrapper : commandWrappers) {
        ExecutionCommand c = wrapper.getExecutionCommand();
        String roleStr = c.getRole();
        HostRoleStatus status = s.getHostRoleStatus(host, roleStr);
        i_my++;
        if (LOG.isTraceEnabled()) {
          LOG.trace("Host task {}) id = {} status = {} (role={}), roleCommand = {}", i_my, c.getTaskId(), status, roleStr, c.getRoleCommand());
        }
        boolean hostDeleted = false;
        if (null != cluster) {
          Service svc = null;
          if (c.getServiceName() != null && !c.getServiceName().isEmpty()) {
            svc = cluster.getService(c.getServiceName());
          }

          ServiceComponent svcComp = null;
          Map<String, ServiceComponentHost> scHosts = null;
          try {
            if (svc != null) {
              svcComp = svc.getServiceComponent(roleStr);
              scHosts = svcComp.getServiceComponentHosts();
            }
          } catch (ServiceComponentNotFoundException scnex) {
            String msg = String.format(
              "%s is not not a service component, assuming its an action",
              roleStr);
            LOG.debug(msg);
          }

          hostDeleted = (scHosts != null && !scHosts.containsKey(host));
          if (hostDeleted) {
            String message = String.format(
              "Host component information has not been found.  Details:" +
                "cluster=%s; host=%s; service=%s; component=%s; ",
              c.getClusterName(), host,
              svcComp == null ? "null" : svcComp.getServiceName(),
              svcComp == null ? "null" : svcComp.getName());
            LOG.warn(message);
          }
        }

        //basic timeout for stage
        long commandTimeout = actionTimeout;
        if (taskTimeoutAdjustment) {
          Map<String, String> commandParams = c.getCommandParams();
          String timeoutKey = ExecutionCommand.KeyNames.COMMAND_TIMEOUT;
          if (commandParams != null && commandParams.containsKey(timeoutKey)) {
            String timeoutStr = commandParams.get(timeoutKey);
            commandTimeout += Long.parseLong(timeoutStr) * 1000; // Converting to milliseconds
          } else {
            LOG.error("Execution command has no timeout parameter" +
              c);
          }
        }

        // Check that service host component is not deleted
        boolean isHostStateUnknown = false;
        if (hostDeleted) {

          String message = String.format(
            "Host not found when trying to schedule an execution command. " +
              "The most probable reason for that is that host or host component " +
              "has been deleted recently. The command has been aborted and dequeued." +
              "Execution command details: " +
              "cmdId: %s; taskId: %s; roleCommand: %s",
            c.getCommandId(), c.getTaskId(), c.getRoleCommand());
          LOG.warn("Host {} has been detected as non-available. {}", host, message);
          // Abort the command itself
          // We don't need to send CANCEL_COMMANDs in this case
          db.abortHostRole(host, s.getRequestId(), s.getStageId(), c.getRole(), message);
          if (c.getRoleCommand().equals(RoleCommand.ACTIONEXECUTE)) {
            processActionDeath(cluster.getClusterName(), c.getHostname(), roleStr);
          }
          status = HostRoleStatus.ABORTED;
        } else if (timeOutActionNeeded(status, s, hostObj, roleStr, now, commandTimeout)
          || (isHostStateUnknown = isHostStateUnknown(s, hostObj, roleStr))) {
          // Process command timeouts
          if (s.getAttemptCount(host, roleStr) >= maxAttempts || isHostStateUnknown) {
            LOG.warn("Host: {}, role: {}, actionId: {} expired and will be failed", host, roleStr,
              s.getActionId());

            // determine if the task should be auto skipped
            boolean isSkipSupported = s.isAutoSkipOnFailureSupported();
            HostRoleCommand hostRoleCommand = s.getHostRoleCommand(c.getTaskId());
            if (isSkipSupported && null != hostRoleCommand) {
              isSkipSupported = hostRoleCommand.isFailureAutoSkipped();
            }

            db.timeoutHostRole(host, s.getRequestId(), s.getStageId(), c.getRole(), isSkipSupported, isHostStateUnknown);
            //Reinitialize status
            status = s.getHostRoleStatus(host, roleStr);

            if (null != cluster) {
              if (!RoleCommand.CUSTOM_COMMAND.equals(c.getRoleCommand())
                && !RoleCommand.SERVICE_CHECK.equals(c.getRoleCommand())
                && !RoleCommand.ACTIONEXECUTE.equals(c.getRoleCommand())) {
                //commands above don't affect host component state (e.g. no in_progress state in process), transition will fail
                transitionToFailedState(cluster.getClusterName(), c.getServiceName(), roleStr, host, now, false);
              }
              if (c.getRoleCommand().equals(RoleCommand.ACTIONEXECUTE)) {
                processActionDeath(cluster.getClusterName(), c.getHostname(), roleStr);
              }
            }

            // Dequeue command
            LOG.info("Removing command from queue, host={}, commandId={} ", host, c.getCommandId());
          } else {
            cancelCommandOnTimeout(Collections.singletonList(s.getHostRoleCommand(host, roleStr)), commandsToEnqueue);

            LOG.info("Host: {}, role: {}, actionId: {} timed out and will be rescheduled", host,
              roleStr, s.getActionId());

            // reschedule command
            commandsToSchedule.add(c);
            LOG.trace("===> commandsToSchedule(reschedule)={}", commandsToSchedule.size());
          }
        } else if (status.equals(HostRoleStatus.PENDING)) {
          // in case of DEPENDENCY_ORDERED stage command can be scheduled only if all of it's dependencies are
          // already finished
          if (CommandExecutionType.STAGE == s.getCommandExecutionType() ||
                (CommandExecutionType.DEPENDENCY_ORDERED == s.getCommandExecutionType() &&
                  CommandExecutionType.DEPENDENCY_ORDERED == configuration.getStageExecutionType() &&
                  areCommandDependenciesFinished(c, s, rolesCommandsInProgress))) {

            //Need to schedule first time
            commandsToSchedule.add(c);
            LOG.trace("===>commandsToSchedule(first_time)={}", commandsToSchedule.size());
          }
        }

        updateRoleStats(status, roleStats.get(roleStr));
        if (status == HostRoleStatus.FAILED) {
          LOG.info("Role {} on host {} was failed", roleStr, host);
        }

      }
    }
    LOG.debug("Collected {} commands to schedule in this wakeup.", commandsToSchedule.size());
    return roleStats;
  }

  /**
   * Returns true if all command dependencies are already finished (not IN_PROGRESS states).
   * @param command
   * @param stage
   * @param rolesCommandsInProgress
   * @return
   */
  private boolean areCommandDependenciesFinished(ExecutionCommand command, Stage stage, Set<RoleCommandPair>
    rolesCommandsInProgress) {
    boolean areCommandDependenciesFinished = true;
    RoleCommandOrder rco = roleCommandOrderProvider.getRoleCommandOrder(stage.getClusterId());
    if (rco != null) {
      RoleCommandPair roleCommand = new
              RoleCommandPair(Role.valueOf(command.getRole()), command.getRoleCommand());
      Set<RoleCommandPair> roleCommandDependencies = rco.getDependencies().get(roleCommand);

      // check if there are any dependencies IN_PROGRESS
      if (roleCommandDependencies != null) {
        // remove eventual references to the same RoleCommand
        roleCommandDependencies.remove(roleCommand);
        if (CollectionUtils.containsAny(rolesCommandsInProgress, roleCommandDependencies)) {
          areCommandDependenciesFinished = false;
        }
      }
    }

    return areCommandDependenciesFinished;

  }

  /**
   * Generate a OPFailed event before aborting all operations in the stage
   * @param stage
   */
  private void abortOperationsForStage(Stage stage) {
    long now = System.currentTimeMillis();

    for (String hostName : stage.getHosts()) {
      List<ExecutionCommandWrapper> commandWrappers =
        stage.getExecutionCommands(hostName);

      for(ExecutionCommandWrapper wrapper : commandWrappers) {
        ExecutionCommand c = wrapper.getExecutionCommand();
        transitionToFailedState(stage.getClusterName(), c.getServiceName(),
                c.getRole(), hostName, now, true);
      }
    }
    Collection<HostRoleCommandEntity> abortedOperations = db.abortOperation(stage.getRequestId());

    for (HostRoleCommandEntity command: abortedOperations) {
      if (command.getRoleCommand().equals(RoleCommand.ACTIONEXECUTE)) {
        String clusterName = stage.getClusterName();
        processActionDeath(clusterName, command.getHostName(), command.getRole().name());
      }
    }
  }

  /**
   * Raise a OPFailed event for a SCH
   * @param clusterName
   * @param serviceName
   * @param componentName
   * @param hostname
   * @param timestamp
   */
  private void transitionToFailedState(String clusterName, String serviceName,
                                       String componentName, String hostname,
                                       long timestamp,
                                       boolean ignoreTransitionException) {

    try {
      Cluster cluster = clusters.getCluster(clusterName);

      ServiceComponentHostOpFailedEvent failedEvent =
        new ServiceComponentHostOpFailedEvent(componentName,
          hostname, timestamp);

      if (serviceName != null && ! serviceName.isEmpty() &&
              componentName != null && ! componentName.isEmpty()) {
        Service svc = cluster.getService(serviceName);
        ServiceComponent svcComp = svc.getServiceComponent(componentName);
        ServiceComponentHost svcCompHost =
                svcComp.getServiceComponentHost(hostname);
        svcCompHost.handleEvent(failedEvent);
      } else {
        LOG.info("Service name is " + serviceName + ", component name is " + componentName +
                "skipping sending ServiceComponentHostOpFailedEvent for " + componentName);
      }

    } catch (ServiceComponentNotFoundException scnex) {
      LOG.debug("{} associated with service {} is not a service component, assuming it's an action.", componentName, serviceName);
    } catch (ServiceComponentHostNotFoundException e) {
      String msg = String.format("Service component host %s not found, " +
              "unable to transition to failed state.", componentName);
      LOG.warn(msg, e);
    } catch (InvalidStateTransitionException e) {
      if (ignoreTransitionException) {
        LOG.debug("Unable to transition to failed state.", e);
      } else {
        LOG.warn("Unable to transition to failed state.", e);
      }
    } catch (AmbariException e) {
      LOG.warn("Unable to transition to failed state.", e);
    }
  }

  /**
   * Populates a map < role_name, role_stats>.
   */
  private Map<String, RoleStats> initRoleStats(Stage s) {
    // Meaning: how many hosts are affected by commands for each role
    Map<Role, Integer> hostCountsForRoles = new HashMap<>();
    // < role_name, rolestats >
    Map<String, RoleStats> roleStats = new TreeMap<>();

    for (String host : s.getHostRoleCommands().keySet()) {
      Map<String, HostRoleCommand> roleCommandMap = s.getHostRoleCommands().get(host);
      for (String role : roleCommandMap.keySet()) {
        HostRoleCommand c = roleCommandMap.get(role);
        if (hostCountsForRoles.get(c.getRole()) == null) {
          hostCountsForRoles.put(c.getRole(), 0);
        }
        int val = hostCountsForRoles.get(c.getRole());
        hostCountsForRoles.put(c.getRole(), val + 1);
      }
    }

    for (Role r : hostCountsForRoles.keySet()) {
      RoleStats stats = new RoleStats(hostCountsForRoles.get(r),
          s.getSuccessFactor(r));
      roleStats.put(r.name(), stats);
    }
    return roleStats;
  }

  /**
   * Checks if ambari-agent was restarted during role command execution
   * @param host the host with ambari-agent to check
   * @param stage the stage
   * @param role the role to check
   * @return {@code true} if ambari-agent was restarted
   */
  protected boolean wasAgentRestartedDuringOperation(Host host, Stage stage, String role) {
    String hostName = host.getHostName();
    long taskStartTime = stage.getHostRoleCommand(hostName, role).getStartTime();
    long lastAgentStartTime = host.getLastAgentStartTime();
    return taskStartTime > 0 && lastAgentStartTime > 0 && taskStartTime <= lastAgentStartTime;
  }

  /**
   * Checks if timeout is required.
   * @param status      the status of the current role
   * @param stage       the stage
   * @param host        the host object; can be {@code null} for server-side tasks
   * @param role        the role
   * @param currentTime the current
   * @param taskTimeout the amount of time to determine timeout
   * @return {@code true} if timeout is needed
   * @throws AmbariException
   */
  protected boolean timeOutActionNeeded(HostRoleStatus status, Stage stage,
      Host host, String role, long currentTime, long taskTimeout) throws
    AmbariException {
    if (( !status.equals(HostRoleStatus.QUEUED) ) &&
        ( ! status.equals(HostRoleStatus.IN_PROGRESS) )) {
      return false;
    }

    // tasks are held in a variety of in-memory maps that require a hostname key
    // host being null is ok - that means it's a server-side task
    String hostName = (null == host) ? null : host.getHostName();

    // If we have other command in progress for this stage do not timeout this one
    if (hasCommandInProgress(stage, hostName)
            && !status.equals(HostRoleStatus.IN_PROGRESS)) {
      return false;
    }
    if (currentTime >= stage.getLastAttemptTime(hostName, role)
        + taskTimeout) {
      return true;
    }
    return false;
  }

  private boolean isHostStateUnknown(Stage stage, Host host, String role) {
    if (null != host &&
      (host.getState().equals(HostState.HEARTBEAT_LOST) || wasAgentRestartedDuringOperation(host, stage, role))) {
      LOG.debug("Abort action since agent is not heartbeating or agent was restarted.");
      return true;
    }
    return false;
  }

  private boolean hasCommandInProgress(Stage stage, String host) {
    List<ExecutionCommandWrapper> commandWrappers = stage.getExecutionCommands(host);
    for (ExecutionCommandWrapper wrapper : commandWrappers) {
      ExecutionCommand c = wrapper.getExecutionCommand();
      String roleStr = c.getRole();
      HostRoleStatus status = stage.getHostRoleStatus(host, roleStr);
      if (status == HostRoleStatus.IN_PROGRESS) {
        return true;
      }
    }
    return false;
  }

  private ListMultimap<String, ServiceComponentHostEvent> formEventMap(Stage s, List<ExecutionCommand> commands) {
    ListMultimap<String, ServiceComponentHostEvent> serviceEventMap = ArrayListMultimap.create();
    for (ExecutionCommand cmd : commands) {
      String hostname = cmd.getHostname();
      String roleStr = cmd.getRole();
      if (RoleCommand.ACTIONEXECUTE != cmd.getRoleCommand()) {
          serviceEventMap.put(cmd.getServiceName(), s.getFsmEvent(hostname, roleStr).getEvent());
      }
    }
    return serviceEventMap;
  }

  private void processHostRole(RequestEntity r, Stage s, ExecutionCommand cmd, List<ExecutionCommand> commandsToStart,
                               List<ExecutionCommand> commandsToUpdate)
    throws AmbariException {
    long now = System.currentTimeMillis();
    String roleStr = cmd.getRole();
    String hostname = cmd.getHostname();

    // start time is -1 if host role command is not started yet
    if (s.getStartTime(hostname, roleStr) < 0) {

      commandsToStart.add(cmd);
      s.setStartTime(hostname,roleStr, now);
      s.setHostRoleStatus(hostname, roleStr, HostRoleStatus.QUEUED);
    }
    s.setLastAttemptTime(hostname, roleStr, now);
    s.incrementAttemptCount(hostname, roleStr);

    String requestPK = r.getRequestId().toString();
    String stagePk = s.getStageId() + "-" + s.getRequestId();

    // Try to get clusterHostInfo from cache
    Map<String, Set<String>> clusterHostInfo = clusterHostInfoCache.getIfPresent(requestPK);

    if (clusterHostInfo == null) {
      Type type = new TypeToken<Map<String, Set<String>>>() {}.getType();
      clusterHostInfo = StageUtils.getGson().fromJson(r.getClusterHostInfo(), type);
      clusterHostInfoCache.put(requestPK, clusterHostInfo);
    }

    cmd.setClusterHostInfo(clusterHostInfo);

    // Try to get commandParams from cache and merge them with command-level parameters
    Map<String, String> commandParams = commandParamsStageCache.getIfPresent(stagePk);

    if (commandParams == null){
      Type type = new TypeToken<Map<String, String>>() {}.getType();
      commandParams = StageUtils.getGson().fromJson(s.getCommandParamsStage(), type);
      commandParamsStageCache.put(stagePk, commandParams);
    }
    Map<String, String> commandParamsCmd = cmd.getCommandParams();
    commandParamsCmd.putAll(commandParams);
    cmd.setCommandParams(commandParamsCmd);

    try {
      Cluster cluster = clusters.getCluster(s.getClusterName());
      if (null != cluster) {
        // Generate localComponents
        for (ServiceComponentHost sch : cluster.getServiceComponentHosts(hostname)) {
          cmd.getLocalComponents().add(sch.getServiceComponentName());
        }
      }
    } catch (ClusterNotFoundException cnfe) {
      // NOP
    }

    // Try to get hostParams from cache and merge them with command-level parameters
    Map<String, String> hostParams = hostParamsStageCache.getIfPresent(stagePk);
    if (hostParams == null) {
      Type type = new TypeToken<Map<String, String>>() {}.getType();
      hostParams = StageUtils.getGson().fromJson(s.getHostParamsStage(), type);
      hostParamsStageCache.put(stagePk, hostParams);
    }
    Map<String, String> hostParamsCmd = cmd.getHostLevelParams();
    hostParamsCmd.putAll(hostParams);
    cmd.setHostLevelParams(hostParamsCmd);

    // change the hostname in the command for the host itself
    cmd.setHostname(hostsMap.getHostMap(hostname));

    commandsToUpdate.add(cmd);
  }

  /**
   * @param requestId request will be cancelled on next scheduler wake up
   * (if it is in state that allows cancellation, e.g. QUEUED, PENDING, IN_PROGRESS)
   * @param reason why request is being cancelled
   */
  public void scheduleCancellingRequest(long requestId, String reason) {
    synchronized (requestsToBeCancelled) {
      requestsToBeCancelled.add(requestId);
      requestCancelReasons.put(requestId, reason);
    }
  }

  /**
   * Aborts all stages that belong to requests that are being cancelled
   */
  private void processCancelledRequestsList() throws AmbariException {
    synchronized (requestsToBeCancelled) {
      // Now, cancel stages completely
      for (Long requestId : requestsToBeCancelled) {
        // only pull back entities that have not completed; pulling back all
        // entities for the request can cause OOM problems on large requests,
        // like those for upgrades
        List<HostRoleCommandEntity> entitiesToDequeue = hostRoleCommandDAO.findByRequestIdAndStatuses(
            requestId, HostRoleStatus.NOT_COMPLETED_STATUSES);

        if (!entitiesToDequeue.isEmpty()) {
          List<HostRoleCommand> tasksToDequeue = new ArrayList<>(entitiesToDequeue.size());
          for (HostRoleCommandEntity hrcEntity : entitiesToDequeue) {
            HostRoleCommand task = hostRoleCommandFactory.createExisting(hrcEntity);
            tasksToDequeue.add(task);
          }

          String reason = requestCancelReasons.get(requestId);
          cancelHostRoleCommands(tasksToDequeue, reason);
        }

        // abort any stages in progress that belong to this request; don't execute this for all stages since
        // that could lead to OOM errors on large requests, like those for
        // upgrades
        List<Stage> stagesInProgress = db.getStagesInProgressForRequest(requestId);
        for (Stage stageInProgress : stagesInProgress) {
          abortOperationsForStage(stageInProgress);
        }
      }

      requestsToBeCancelled.clear();
      requestCancelReasons.clear();
    }
  }

  /**
   * Cancels host role commands (those that are not finished yet).
   * Dequeues host role commands that have been added to ActionQueue,
   * and automatically generates and adds to ActionQueue CANCEL_COMMANDs
   * for all hostRoleCommands that have already been sent to an agent for
   * execution.
   * @param hostRoleCommands a list of hostRoleCommands
   * @param reason why the request is being cancelled
   */
  void cancelHostRoleCommands(Collection<HostRoleCommand> hostRoleCommands, String reason) throws AmbariException {
    for (HostRoleCommand hostRoleCommand : hostRoleCommands) {
      // There are no server actions in actionQueue
      if (!Role.AMBARI_SERVER_ACTION.equals(hostRoleCommand.getRole())) {
        if (hostRoleCommand.getStatus() == HostRoleStatus.QUEUED ||
              hostRoleCommand.getStatus() == HostRoleStatus.IN_PROGRESS) {
          CancelCommand cancelCommand = new CancelCommand();
          cancelCommand.setTargetTaskId(hostRoleCommand.getTaskId());
          cancelCommand.setReason(reason);
          agentCommandsPublisher.sendAgentCommand(hostRoleCommand.getHostId(), cancelCommand);
        }
      }

      if (hostRoleCommand.getStatus().isHoldingState()) {
        db.abortHostRole(hostRoleCommand.getHostName(),
            hostRoleCommand.getRequestId(),
            hostRoleCommand.getStageId(), hostRoleCommand.getRole().name());
      }

      // If host role is an Action, we have to send an event
      if (hostRoleCommand.getRoleCommand().equals(RoleCommand.ACTIONEXECUTE)) {
        String clusterName = hostRoleCommand.getExecutionCommandWrapper().getExecutionCommand().getClusterName();
        processActionDeath(clusterName,
                hostRoleCommand.getHostName(),
                hostRoleCommand.getRole().name());
      }
    }
  }

  void cancelCommandOnTimeout(Collection<HostRoleCommand> hostRoleCommands, Multimap<Long, AgentCommand> commandsToEnqueue) {
    for (HostRoleCommand hostRoleCommand : hostRoleCommands) {
      // There are no server actions in actionQueue
      if (!Role.AMBARI_SERVER_ACTION.equals(hostRoleCommand.getRole())) {
        if (hostRoleCommand.getStatus() == HostRoleStatus.QUEUED ||
              hostRoleCommand.getStatus() == HostRoleStatus.IN_PROGRESS) {
          CancelCommand cancelCommand = new CancelCommand();
          cancelCommand.setTargetTaskId(hostRoleCommand.getTaskId());
          cancelCommand.setReason("Stage timeout");
          commandsToEnqueue.put(hostRoleCommand.getHostId(), cancelCommand);
        }
      }
    }
  }

  /**
   * Attempts to process kill/timeout/abort of action and send
   * appropriate event to all listeners
   */
  private void processActionDeath(String clusterName,
                                  String hostname,
                                  String role) {
    try {
      // Usually clusterId is defined (except the awkward case when
      // "Distribute repositories/install packages" action has been issued
      // against a concrete host without binding to a cluster)
      Long clusterId = clusterName != null ?
              clusters.getCluster(clusterName).getClusterId() : null;
      CommandReport report = new CommandReport();
      report.setRole(role);
      report.setStdOut("Action is dead");
      report.setStdErr("Action is dead");
      report.setStructuredOut("{}");
      report.setExitCode(1);
      report.setStatus(HostRoleStatus.ABORTED.toString());
      ActionFinalReportReceivedEvent event = new ActionFinalReportReceivedEvent(
              clusterId, hostname, report, true);
      ambariEventPublisher.publish(event);
    } catch (AmbariException e) {
      LOG.error(String.format("Can not get cluster %s", clusterName), e);
    }
  }

  private void updateRoleStats(HostRoleStatus status, RoleStats rs) {
    switch (status) {
      case COMPLETED:
        rs.numSucceeded++;
        break;
      case FAILED:
        rs.numFailed++;
        break;
      case QUEUED:
        rs.numQueued++;
        break;
      case PENDING:
        rs.numPending++;
        break;
      case TIMEDOUT:
        rs.numTimedOut++;
        break;
      case ABORTED:
        rs.numAborted++;
        break;
      case IN_PROGRESS:
        rs.numInProgress++;
        break;
      case HOLDING:
      case HOLDING_FAILED:
      case HOLDING_TIMEDOUT:
        rs.numHolding++;
        break;
      case SKIPPED_FAILED:
        rs.numSkipped++;
        break;
      default:
        LOG.error("Unknown status " + status.name());
    }
  }

  public void setTaskTimeoutAdjustment(boolean val) {
    taskTimeoutAdjustment = val;
  }

  ServerActionExecutor getServerActionExecutor() {
    return serverActionExecutor;
  }

  /**
   * Handles {@link EntityManagerCacheInvalidationEvent} instances and instructs
   * the thread running this scheduler to evict instances from the
   * {@link EntityManager}.
   *
   * @param event
   *          the event to handle (not {@code null}).
   */
  @Subscribe
  public void onEvent(EntityManagerCacheInvalidationEvent event) {
    try {
      if (null != threadEntityManager && threadEntityManager.isOpen()) {
        threadEntityManager.clear();
      }
    } catch (Throwable throwable) {
      LOG.error("Unable to clear the EntityManager for the scheduler thread", throwable);
    }
  }

  static class RoleStats {
    int numInProgress;
    int numQueued = 0;
    int numSucceeded = 0;
    int numFailed = 0;
    int numTimedOut = 0;
    int numPending = 0;
    int numAborted = 0;
    int numHolding = 0;
    int numSkipped = 0;

    final int totalHosts;
    final float successFactor;

    RoleStats(int total, float successFactor) {
      totalHosts = total;
      this.successFactor = successFactor;
    }

    /**
     * Role successful means the role is successful enough to
     */
    boolean isSuccessFactorMet() {
      int minSuccessNeeded = (int) Math.ceil(successFactor * totalHosts);
      return minSuccessNeeded <= numSucceeded;
    }

    private boolean isRoleInProgress() {
      return numPending + numQueued + numInProgress + numHolding > 0;
    }

    /**
     * Role failure means role is no longer in progress and success factor is
     * not met.
     */
    boolean isRoleFailed() {
      return !(isRoleInProgress() || isSuccessFactorMet());
    }

    @Override
    public String toString() {
      StringBuilder builder = new StringBuilder();
      builder.append("numQueued=").append(numQueued);
      builder.append(", numInProgress=").append(numInProgress);
      builder.append(", numSucceeded=").append(numSucceeded);
      builder.append(", numFailed=").append(numFailed);
      builder.append(", numTimedOut=").append(numTimedOut);
      builder.append(", numPending=").append(numPending);
      builder.append(", numAborted=").append(numAborted);
      builder.append(", numSkipped=").append(numSkipped);
      builder.append(", totalHosts=").append(totalHosts);
      builder.append(", successFactor=").append(successFactor);
      return builder.toString();
    }
  }
}


text:
<?php

namespace AppBundle\Entity;

use Doctrine\ORM\Mapping as ORM;
use Symfony\Component\Validator\Constraints as Assert;
use Gedmo\Mapping\Annotation as Gedmo;
use Doctrine\ORM\Mapping\Column;
use Doctrine\Common\Collections\Collection;
use Doctrine\Common\Collections\ArrayCollection;

/**
 * @author Jordan
 *
 * @ORM\Entity(repositoryClass="AppBundle\Repository\TaskRepository")
 * @ORM\Table(name="t_task")
 */
class Task {
	/**
	 * @var integer
	 *
	 * @ORM\Id
	 * @ORM\Column(type="integer")
	 * @ORM\GeneratedValue(strategy="AUTO")
	 */
	private $id;

	/**
	 * @var string
	 *
	 * @ORM\Column(type="string", name="label")
	 * @Assert\NotBlank()
	 */
	private $label;

	/**
	 * @var string
	 *
	 * @ORM\Column(type="text", name="description")
	 */
	private $description;

	/**
	 * @var \DateTime
	 *
	 * @ORM\Column(type="date", name="due_date")
	 */
	private $dueDate;

	/**
	 * @var float
	 *
	 * @ORM\Column(type="float", scale=2, name="duration")
	 */
	private $duration;

	/**
	 * @var RefDurationType
	 *
	 * @ORM\ManyToOne(targetEntity="AppBundle\Entity\RefDurationType")
	 * @ORM\JoinColumn(name="ref_duration_type_id", referencedColumnName="id")
	 */
	private $durationType;

	/**
	 * @var RefPriority
	 *
	 * @ORM\ManyToOne(targetEntity="AppBundle\Entity\RefPriority")
	 * @ORM\JoinColumn(name="ref_priority_id", referencedColumnName="id")
	 */
	private $priority;

	/**
	 * @var RefTaskState
	 *
	 * @ORM\ManyToOne(targetEntity="AppBundle\Entity\RefTaskState")
	 * @ORM\JoinColumn(name="ref_task_state_id", referencedColumnName="id")
	 */
	private $state;

	/**
	 * @var Project
	 *
	 * @ORM\ManyToOne(targetEntity="AppBundle\Entity\Project")
	 * @ORM\JoinColumn(name="project_id", referencedColumnName="id")
	 */
	private $project;

	/**
	 * @var Collection
	 *
	 * @ORM\ManyToMany(targetEntity="AppBundle\Entity\Skill")
	 * @ORM\JoinTable(name="l_task_skill",
	 * 		joinColumns={@ORM\JoinColumn(name="task_id", referencedColumnName="id")},
	 *      inverseJoinColumns={@ORM\JoinColumn(name="skill_id", referencedColumnName="id")}
	 * )
	 */
	private $requiredSkills;

	/**
	 * @var Collection
	 *
	 * @ORM\OneToMany(targetEntity="AppBundle\Entity\RealizationPeriod", mappedBy="task")
	 */
	private $realizationPeriods;

	/**
	 * @var Collection
	 *
	 * @ORM\OneToMany(targetEntity="AppBundle\Entity\Document", mappedBy="task")
	 */
	private $documents;

	/**
	 * @var Project
	 *
	 * @ORM\ManyToOne(targetEntity="AppBundle\Entity\Company")
	 * @ORM\JoinColumn(name="company_id", referencedColumnName="id")
	 * @Assert\NotNull()
	 */
	private $company;

	/**
	 * @var User
	 *
	 * @Gedmo\Blameable(on="create")
	 * @ORM\ManyToOne(targetEntity="AppBundle\Entity\User")
	 * @ORM\JoinColumn(name="created_by", referencedColumnName="id")
	 */
	private $createdBy;

	/**
	 * @var \DateTime
	 *
	 * @Gedmo\Timestampable(on="create")
	 * @ORM\Column(type="datetime", name="created_at")
	 */
	private $createdAt;

	/**
	 * @var User
	 *
	 * @Gedmo\Blameable(on="update")
	 * @ORM\ManyToOne(targetEntity="AppBundle\Entity\User")
	 * @ORM\JoinColumn(name="updated_by", referencedColumnName="id")
	 */
	private $updatedBy;

	/**
	 * @var \DateTime
	 *
	 * @Gedmo\Timestampable(on="update")
	 * @ORM\Column(type="datetime", name="updated_at")
	 */
	private $updatedAt;

	public function __construct() {
		$this->requiredSkills = new ArrayCollection();
	}

	public function getId() {
		return $this->id;
	}
	public function getLabel() {
		return $this->label;
	}
	public function setLabel($label) {
		$this->label = $label;
		return $this;
	}
	public function getDescription() {
		return $this->description;
	}
	public function setDescription($description) {
		$this->description = $description;
		return $this;
	}
	public function getDueDate() {
		return $this->dueDate;
	}
	public function setDueDate(\DateTime $dueDate) {
		$this->dueDate = $dueDate;
		return $this;
	}
	public function getDuration() {
		return $this->duration;
	}
	public function setDuration($duration) {
		$this->duration = $duration;
		return $this;
	}
	public function getDurationType() {
		return $this->durationType;
	}
	public function setDurationType(RefDurationType $durationType) {
		$this->durationType = $durationType;
		return $this;
	}
	public function getPriority() {
		return $this->priority;
	}
	public function setPriority(RefPriority $priority) {
		$this->priority = $priority;
		return $this;
	}
	public function getState() {
		return $this->state;
	}
	public function setState(RefTaskState $state) {
		$this->state = $state;
		return $this;
	}
	public function getProject() {
		return $this->project;
	}
	public function setProject(Project $project) {
		$this->project = $project;
		return $this;
	}
	public function getCreatedBy() {
		return $this->createdBy;
	}
	public function setCreatedBy(User $createdBy) {
		$this->createdBy = $createdBy;
		return $this;
	}
	public function getCreatedAt() {
		return $this->createdAt;
	}
	public function setCreatedAt(\DateTime $createdAt) {
		$this->createdAt = $createdAt;
		return $this;
	}
	public function getUpdatedBy() {
		return $this->updatedBy;
	}
	public function setUpdatedBy(User $updatedBy) {
		$this->updatedBy = $updatedBy;
		return $this;
	}
	public function getUpdatedAt() {
		return $this->updatedAt;
	}
	public function setUpdatedAt(\DateTime $updatedAt) {
		$this->updatedAt = $updatedAt;
		return $this;
	}
	public function getRequiredSkills() {
		return $this->requiredSkills;
	}
	public function setRequiredSkills(Collection $requiredSkills) {
		$this->requiredSkills = $requiredSkills;
		return $this;
	}
	public function addRequiredSkill(Skill $skill) {
		if (!$this->requiredSkills->contains($skill)) {
			$this->requiredSkills->add($skill);
		}
	}
	public function removeRequiredSkill(Skill $skill) {
		if ($this->requiredSkills->contains($skill)) {
			$this->requiredSkills->removeElement($skill);
		}
	}
	public function getDocuments() {
		return $this->documents;
	}
	public function setDocuments(Collection $documents) {
		$this->documents = $documents;
		return $this;
	}
	public function addDocument(Document $document) {
		if (!$this->documents->contains($document)) {
			$this->documents->add($document);
		}
	}
	public function removeDocument(Document $document) {
		if ($this->documents->contains($document)) {
			$this->documents->removeElement($document);
		}
	}
	public function getCompany() {
		return $this->company;
	}
	public function setCompany(Project $company) {
		$this->company = $company;
		return $this;
	}
	public function getRealizationPeriods() {
		return $this->realizationPeriods;
	}
	public function setRealizationPeriods(Collection $realizationPeriods) {
		$this->realizationPeriods = $realizationPeriods;
		return $this;
	}
	public function addRealizationPeriod(RealizationPeriod $realizationPeriod) {
		if (!$this->realizationPeriods->contains($realizationPeriod)) {
			$this->realizationPeriods->add($realizationPeriod);
		}
	}
	public function removeRealizationPeriod(RealizationPeriod $realizationPeriod) {
		if ($this->realizationPeriods->contains($realizationPeriod)) {
			$this->realizationPeriods->removeElement($realizationPeriod);
		}
	}
}