text:
<?php namespace Drivers\Database\MySQL;

use Drivers\Utilities\ArrayUtility;

class MySQLQuery {

	/**
	 *@var object MySQL connection instance object
	 *@readwrite
	 */
	protected $connector;

	/**
	 *@var
	 *@read
	 */
	protected $froms;

	/**
	 *@var
	 *@read
	 */
	protected $fields;

	/**
	 *@var
	 *@read
	 */
	protected $limits;

	/**
	 *@var
	 *@read
	 */
	protected $offset;

	/**
	 *@var
	 *@read
	 */
	protected $orders;

	/**
	 *@var
	 *@read
	 */
	protected $distinct = ' ';

	/**
	 *@var
	 *@read
	 */
	protected $directions;

	/**
	 *@var
	 *@read
	 */
	protected $joins = array();

	/**
	 *@var
	 *@read
	 */
	protected $wheres = array();

	/**
	 *This stores the database connection instance object
	 *
	 **@param object $instance This is the database connection object instance
	 */
	public function __construct(array $instance)
	{
		//assign the connection object instance to the $this->connnector variable
		$this->connector = $instance['connector'];

	}

	/**
	 *This method quotes input data according to how MySQL woudl expect it
	 *
	 *@param mixed String|Array Input to be quoted
	 *@return mixed Output after input is quoted
	 */
	protected function quote($value)
	{
		//if string, quote and return
		if ( is_string($value) || is_int($value) )
		{
			//call connector method to escape string
			$escaped = $this->connector->escape($value);

			//return escaped string
			return "'{$escaped}'";

		}

		//if input is array, loop through values escaping
		elseif ( is_array($value) )
		{
			//define array to contain new quoted values
			$buffer = array();

			//loop through the array one item at a time
			foreach ($value as $i)
			{
				//quote string and add to the $buffer array container
				array_push($buffer,$this->quote($i));

				//join the array elements to form one string
				$buffer = join(", ", $buffer);

				//quote new string, put in parenthesis and return
				return "({$buffer})";

			}

		}

		//return null if null value was passed
		elseif ( is_null($value) )
		{
			//return null as string
			return 'NULL';

		}

		//if boolean, return interger value of boolean
		elseif ( is_bool($value) )
		{
			//get interger value and return
			return (int)$value;

		}

		//check if empty value was returned
		elseif ( empty($value) )
		{
			//return empty string
			return "' '";

		}

		//if input value does not fall among any of these, escape and return output
		else return $this->connector->escape($value);

	}

	/**
	 *
	 *
	 */
	public function from($from, $fields = array("*"))
	{
		//check if from is empty
		if ( empty($from))
		{
			//throw new exception
			throw new MySQLException("Invalid argument passed to from clause", 1);

		}

		//set the protected $from value
		$this->froms = $from;

		//check if fields is sset
		if($fields)
		{
			//set the fields param
			$this->fields[$from] = $fields;

		}

		//return this object instance
		return $this;

	}

	/**
	 *
	 *
	 */
	public function join($join, $table, $on, $fields = array())
	{
		//throw exception if $join passed is empty
		if ( empty($join) ||  empty($table) || empty($on) )
		{
			//throw exception for invalid argument
			throw new MySQLException("Invalid argument $join passed for the Join Clause", 1);

		}

		//throw exception if the $on passed is empty
		if ( empty($on) )
		{
			//throw exception
			throw new MySQLException("Invalid argument $on passed for the Join Clause", 1);

		}

		//add to the fields property
		$this->fields += array($table => $fields);

		//populate the joins property
		$this->joins[] = strtoupper($join) . " JOIN {$table} ON {$on}";

		//return instance of this object
		return $this;

	}

	/**
	 *
	 *
	 *
	 */
	public function limit($limit, $page = 1)
	{
		if ( empty($limit) )
		{
			//throw exception if value of limit passed is negative
			throw new MySQLException("Empty argument passed for $limit in method limit()", 1);

		}

		//set the limits property
		$this->limits = $limit;

		//set the offset property
		$this->offset = (int)$limit * ($page - 1);

		//return this object instance
		return $this;

	}
	/**
	 *
	 *
	 *
	 */
	public function unique()
	{

		//set the value of distinct
		$this->distinct = ' DISTINCT ';

		//return this object instance
		return $this;

	}

	/**
	 *
	 *
	 */
	public function order($order, $direction = 'asc')
	{
		//throw esception if empty value for $order was passed
		if ( empty($order) )
		{
			//throw exception
			throw new MySQLException("Empty value passed for parameter $order in order() method", 1);

		}

		//set the orders property
		$this->orders = $order;

		//set the directions property
		$this->directions = $direction;

		return $this;

	}

	/**
	 *This method defines the where parameters of the query string
	 *
	 *@param null
	 *@return object Sets the where parameter and returns this instance of the query object
	 */
	public function where()
	{
		//get the arguments
		$arguments = func_get_args();

		//throw exception if argument pairs do ot match
		if ( is_float( sizeof($arguments) / 2 ) )
		{
			//throw exception
			throw new MySQLException("No arguments passed for the where clause");

		}

		//check if only one argument pair was passed'
		if ( sizeof($arguments) == 2)
		{
			//get first argument and replace ? with a placeholder
			$arguments[0] = preg_replace("#\?#", "%s", $arguments[0]);

			//quote the argument content
			$arguments[1] = $this->quote($arguments[1]);

			//populate the wheres array
			$this->wheres[] = call_user_func_array("sprintf", $arguments);

			//return this object instance
			return $this;

		}

		//there are multiple argument supplied
		else
		{
			//get the number of iterations
			$count = sizeof($arguments) / 2;

			//loop through number of iterations population the wheres array
			for ($i = 0; $i < $count; $i++ )
			{
				//get the batch of array to work with
				$argumentsPair  = array_splice($arguments, 0, 2);

				//get first argument and replace ? with a placeholder
				$argumentsPair[0] = preg_replace("#\?#", "%s", $argumentsPair[0]);

				//quote the argument content
				$argumentsPair[1] = $this->quote($argumentsPair[1]);

				//populate the wheres array
				$this->wheres[] = call_user_func_array("sprintf", $argumentsPair);

			}

			//return this object instance
			return $this;

		}

	}

	/**
	 *This method builds a select query string
	 *
	 *@param null
	 *@return string The select query string formed
	 */
	protected function buildSelect()
	{
		//define a container fields array
		$fields = array();

		//
		$where = $order = $limit = $join ="";

		//
		$template = "SELECT %s %s FROM %s %s %s %s %s";

		//loop through the fields
		foreach ($this->fields as $table => $tableFields)
		{
			//loop through the fields checking for aliases
			foreach ($tableFields as $field => $alias)
			{
				//if field is string, return field name
				if ( is_string($field) && $field != 'COUNT(1)')
				{
					//add to fields array
					$fields[] = "{$table}.{$field} AS {$alias}";

				}
				elseif (is_string($field) && $field == 'COUNT(1)')
				{
					//add to fields array
					$fields[] = "{$field} AS {$alias}";

				}
				//not an array, return alias
				else
				{
					//get alias as field name
					$fields[] = "{$table}.{$alias}";

				}

			}

		}

		//turn the fields into a string
		$fields = join(", ", $fields);

		//get the query joins
		$queryJoin = $this->joins;

		//check if the join parameter is empty
		if ( ! empty($queryJoin) )
		{
			//add the joins
			$join = join(" ", $queryJoin);

		}

		//get the where clause
		$queryWhere = $this->wheres;

		//check if the wheres clause is empty
		if ( ! empty($queryWhere) )
		{
			//append where clause
			$joined = join(" AND ", $queryWhere);

			//assign string to $where var
			$where = "WHERE {$joined}";

		}

		//get the order clause
		$queryOrder = $this->orders;

		//check if the orders clause is empty
		if ( ! empty($queryOrder) )
		{
			//get the direction if order
			$orderDirection = $this->directions;

			//assign string to order variable
			$order = "ORDER BY {$queryOrder} {$orderDirection}";

		}

		//get the limits clause
		$queryLimit = $this->limits;

		//check if the limits clause is empty
		if ( ! empty($queryLimit) )
		{
			//get the offset
			$limitOffset = $this->offset;

			//if offset has been defined
			if ( $limitOffset )
			{
				//get the limit string with offset
				$limit = "LIMIT {$limitOffset}, {$queryLimit}";

			}
			//limit offset undefined
			else
			{
				//get the limit string
				$limit = "LIMIT {$queryLimit}";

			}

		}

		//return the formated query string in the format of $template
		return sprintf($template, $this->distinct, $fields, $this->froms, $join, $where, $order, $limit);

	}

	/**
	 *This method builds the query string for inserting one row of records into the database
	 *
	 *@param array The row of data to be inserted in associative array
	 *@return string The formed insert query string
	 */
	protected function buildInsert($data)
	{
		//define a fields container array
		$fields = array();

		//define a values container array
		$values = array();

		//define a template format for query
		$template = "INSERT INTO %s (%s) VALUES (%s)";

		//loop through the input data
		foreach ($data as $field => $value)
		{
			//populate the fields array
			$fields[] = $field;

			//populate the values array
			$values[] = $this->quote($value);

		}

		//convert strings array to string
		$fields = join(", ", $fields);

		//convert values array to string
		$values = join(", ", $values);

		//return the formated query string
		return sprintf($template, $this->froms, $fields, $values);

	}
	/**
	 *This method builds the insert query for more than one row of data
	 *
	 *@param array The data to be inserted in associative array
	 *@return string The formed query string for this insert operation
	 */
	protected function buildBulkInsert($data)
	{
		//define a fields container array
		$fields = array();

		//define a values container array
		$values = array();

		//define a template format for query
		$template = "INSERT INTO %s (%s) VALUES %s";

		//get the fields array
		$fieldsArray = $data[0];

		//get the fields
		foreach ($fieldsArray as $field => $value)
		{
			//populate the fields array
			$fields[] = $field;

		}
		//get the count of number of rows
		$count = sizeof($data);

		//loop through the input data composing input data
		for ( $i = 0; $i < $count; $i++ )
		{
			//remove the first array and return
			$array = $data[$i];

			//define array to contain the string components
			$valuesArray = array();

			//loop through the array composing the values
			foreach ($array as $field => $value)
			{
				//populate the values array
				$valuesArray[] = $this->quote($value);

			}

			//convert strings array to string
			$values[] = '(' . join(", ", $valuesArray) . ')';

		}

		//convert strings array to string
		$fields = join(", ", $fields);

		//convert values array to string
		$values = join(", ", $values);

		//return the formated query string
		return sprintf($template, $this->froms, $fields, $values);

	}

	/**
	 *This method builds an update query for a single record of data
	 *
	 *@param array The data to be updated into the database
	 *@return string The forrmed SQL query string for this update operation
	 */
	protected function buildUpdate($data)
	{
		//define the parts container array
		$parts = array();

		//define the continer vars initializing to empty
		$where = $limit = '';

		//define the template format string
		$template = "UPDATE %s SET %s %s %s";

		//loop through the input data array, populating the parts array
		foreach ($data as $field => $value)
		{
			//populate the parts array
			$parts[] = "{$field}=" . $this->quote($value);

		}

		//convert parts array into string
		$parts = join(", ", $parts);

		//get the where clause
		$queryWhere = $this->wheres;

		//check if where clause is empty
		if ( ! empty($queryWhere) )
		{
			//convert where clause array to string
			$joined = join(", ", $queryWhere);

			//set the where var valus
			$where = "WHERE {$joined}";

		}

		//get the limits var value
		$queryLimit = $this->limits;

		//check if the limit clause is empty
		if ( ! empty($queryLimit) )
		{
			//get the offset value
			$limitOffset = $this->offset;

			//compose limit string
			$limit = "LIMIT {$queryLimit} {$limitOffset}";

		}

		//return the formated query string
		return sprintf($template, $this->froms, $parts, $where, $limit);

	}
	/**
	 *This method builds the SQL query string for updating large amounts of data
	 *
	 *
	 *@param array The data to be updated in multidimensional array
	 *@param array The field names to be updated in a numeric array
	 *@param array The unique field ids to use for updating in numberic array
	 *@return string The formed SQL update query string
	 */
	protected function buildBulkUpdate($data, $fields, $ids, $key)
	{

		//define the parts container array
		$parts = array();

		//define the template format string
		$template = "UPDATE %s SET %s WHERE %s IN (%s) ";

		//loop through the fields array composing the array
		foreach ($fields as $index => $field )
		{
			//initialize the subparts variable
			$subparts = $field . ' = (CASE ' . $key . ' ';

			//loop through the data array composing parts
			foreach ($data as $id => $info )
			{
	            //check if array is not empty
	            if ( ! empty($info) )
	            {
					//echo "<pre>";print_r($info);exit();
					$subparts .=  ' WHEN '. $id . ' THEN ' . $this->quote($info[$field]) . ' ';

	            }

			}

			//finish the subpart
			$subparts .= ' END) ';

			//add this to the main parts array
			$parts[] = $subparts;

		}

		//convert parts array into string
		$parts = join(", ", $parts);

		//get the where clause
		$queryWhere = $ids;

		//check if where clause is empty
		if ( ! empty($queryWhere) )
		{
			//convert where clause array to string
			$where = join(", ", $queryWhere);

		}

		//return the formated query string
		return sprintf($template, $this->froms, $parts, $key, $where);

	}

	/**
	 *This method builds the SQL query string to perform a delete query
	 *
	 *@param null
	 *@return string The SQL query string for delete operation.
	 */
	protected function buildDelete()
	{
		//initialize $where and $limit container vars to empty strings
		$where = $limit = '';

		//define the string formating template
		$template = "DELETE FROM %s %s %s";

		//get the where clause variable
		$queryWhere = $this->wheres;

		//check if where clause is empty
		if ( ! empty($queryWhere) )
		{
			//convert where clause array to string
			$joined = join(", ", $queryWhere);

			//compose the where string
			$where = "WHERE {$joined}";

		}

		//get the limit clause variable
		$queryLimit = $this->limits;

		//check if limit clause is empty
		if ( ! empty($queryLimit) )
		{
			//get the offset value
			$limitOffset = $this->offset;

			//compose the limit clause string
			$limit = "LIMIT {$queryLimit} {$limitOffset}";

		}

		//return the formated query string
		return sprintf($template, $this->froms, $where, $limit);

	}

	/**
	 *This methods inserts/updates one row of data into the database
	 *
	 *@param array The array containing the data to be inserted
	 *@return bool true Returns true if the query execution was successful
	 */
	public function save($data)
	{
		//get the size of the wheres parameter
		$doInsert = sizeof($this->wheres) == 0;

		//check if doInsert is true
		if ( $doInsert )
		{
			//get insert query string
			$sql = $this->buildInsert($data);

		}

		//not insert, this should be an update
		else
		{
			//get update query string
			$sql = $this->buildUpdate($data);

		}
//echo $sql . '<pre>';print_r($data); exit();
		//excecute query
		$result = $this->connector->execute($sql);

		//check if query execution failure
		if ( $result === false)
		{
			//throw exception
			throw new MySQLException($this->connector->lastError());

		}

		//if this was an insert, get the insert id
		if( $doInsert )
		{
			//get last insert id and retunr
			return $this->connector->lastInsertId();
		}

		//if this was update and sucess, return 0 to show operation successful
		return true;

	}
	/**
	 *The method perform insert/update of large amounts of data
	 *
	 *@param array The data to be inserted/updated in a multidimensional array
	 *@param array The fields into which the data is to be inerted ot updated
	 *@param array For update query, The unique id fields to use for updating
	 *@return boolean true Return 0 when query execution is success to indicate true
	 */
	public function saveBulk($data, $fields = null , $ids = null, $key = null )
	{
		//get the size of the wheres parameter
		$doInsert = sizeof($this->wheres) == 0;

		//check if doInsert is true
		if ( $doInsert )
		{
			//get insert query string
			$sql = $this->buildBulkInsert($data);

		}

		//not insert, this should be an update
		else
		{
			//get update query string
			$sql = $this->buildBulkUpdate($data, $fields, $ids, $key);
		}

		//excecute query
		$result = $this->connector->execute($sql);

		//check if query execution failure
		if ( $result === false)
		{
			//throw exception
			throw new MySQLException($this->connector->lastError());

		}

		//if this was an insert, get the insert id
		if( $doInsert )
		{
			//get last insert id and retunr
			return $this->connector->lastInsertId();
		}

		//if this was update and sucess, return 0 to show operation successful
		return 0;

	}

	/**
	 *This method deletes a set of rows that match the query parameters provided
	 *
	 *@param null
	 *@return array The affected rows by this delete action in an array
	 */
	public function delete()
	{
		//build the delete query string
		$sql = $this->buildDelete();

		//execute the query string
		$result = $this->connector->execute($sql);

		//throw error if there was an error perfrorming this query
		if ( $result === false )
		{
			//throw excepton
			throw new MySQLException();

		}

		//if delete was successfull, get and return the number of affected rows
		return $this->connector->affectedRows();

	}

	/**
	 *This method returns the first row match in a query
	 *
	 *@param null
	 *@return array The first element found in an array
	 */
	public function first()
	{
		//get the limit clause value
		$limit = $this->limits;

		//get the offset value
		$limitOffset = $this->offset;

		//set the limit to 1
		$this->limit(1);

		//get all
		$all = $this->all();

		//get the first
		$first = ArrayUtility::first($all);

		//if limit is defined
		if ( $limit )
		{
			//set the limit property
			$this->limits = $limit;

		}

		//if limit offset is set
		if ( $limitOffset )
		{
			//set the offset
			$this->offset = $limitOffset;

		}

		//return the first query row
		return $first;

	}

	/**
	 *This me`thod counts the number of rows returned per query
	 *
	 *
	 *@param null
	 *@return int The count of the number of rows
	 */
	public function count()
	{
		//get the limit clause value
		$limit = $this->limits;

		//get the offset value
		$limitOffset = $this->offset;

		//get the fields clause
		$fields = $this->fields;

		//compose the fields array
		$this->fields = array($this->froms => array("COUNT(1)" => "rows"));

		//set limit to 1
		$this->limit(1);

		//get the first row
		$row = $this->first();

		//set the value of the fields
		$this->fields = $fields;

		//check if fields is defined
		if ( $fields )
		{
			//set the value o fields
			$this->fields = $fields;

		}

		//check if the limit value is set
		if ( $limit )
		{
			//set the value of limit
			$this->limits = $limit;

		}

		//check if the limit offset value is defined
		if ( $limitOffset )
		{
			//set the offset value
			$this->offset = $limitOffset;

		}

		//return the rows count
		return $row[0]['rows'];

	}

	/**
	 *This method returns all rows that match the query parameters
	 *
	 *@param null
	 *@return array The query results in array format
	 */
	public function all()
	{
		//build the select query
		$sql = $this->buildSelect();

		//execute the sql query
		$result = $this->connector->execute($sql);

		//check if the query return an error and throw exception
		if ( $result === false )
		{
			//get the erro message
			$error = $this->connector->lastError();

			//throw exception
			throw new MySQLException("There was an error with your SQL Query : {$error}");

		}

		//define container rows() array
		$rows = array();

		//loop through the result object composing the rows array
		for( $i = 0; $i < $result->num_rows; $i++ )
		{
			//add to the rows array
			$rows[] = $result->fetch_array(MYSQLI_ASSOC);

		}

		//return the full rows array
		return $rows;
	}

}


text:
package org.apache.tapestry5.cdi.test.pages;

import javax.inject.Inject;

import org.apache.tapestry5.cdi.test.beans.Soup;

public class RequestScopePage {

	@Inject
	private Soup soup1;

	@Inject
	private Soup soup2;

	 public String getRequestScopePojo(){

	    	if(soup1 !=null){
	    		return "request:" + soup1.getCheckNames();
	    	}
	    	return "";

	    }
}

redpajama_set_name: RedPajamaGithub


text:
package com.mycom.bds.zk.util;

import java.io.File;
import java.text.DateFormat;
import java.text.SimpleDateFormat;
import java.util.Date;

import org.apache.zookeeper.ZooKeeper;

public class MyZKUtils {

    private static DateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm");

    /**
     * @param parent
     * @param fileName
     * @return file path with forward slash
     */
    public static String getPath(File parent, String fileName) {
        File file = new File(parent, fileName);
        ensureExists(file);
        String absolutePath = file.getAbsolutePath();
        absolutePath = absolutePath.replace("\\", "/");
        System.out.println("path is " + absolutePath);
        return absolutePath;
    }

    /**
     * throws {@link RuntimeException} if file does not exist
     */
    public static void ensureExists(File file) {
        if (!file.exists()) {
            throw new RuntimeException("File '" + file.getAbsolutePath() + "' does not exist.");
        }
    }

    public static byte[] data() {
        return "test date".getBytes();
    }

    public static void print(String message) {
        String dateMessage = dateFormat.format(new Date());
        System.out.println(dateMessage + "::" + message);

    }

    public static String getConnectedServer(ZooKeeper zk) {
        String result = "remoteserver:None";
        String connectionString = zk.toString();
        int indexOfRemoteserver = connectionString.indexOf("remoteserver:");
        if (-1 == indexOfRemoteserver) {
            return result;
        }
        int indexOfSpace = connectionString.indexOf(" ", indexOfRemoteserver);
        if (-1 == indexOfSpace) {
            return result;
        }
        return connectionString.substring(indexOfRemoteserver, indexOfSpace + 1);
    }
}


text:
package org.apache.spark

import java.io.File
import java.net.{MalformedURLException, URI}
import java.nio.charset.StandardCharsets
import java.nio.file.{Files => JavaFiles}
import java.nio.file.attribute.PosixFilePermission._
import java.util.EnumSet
import java.util.concurrent.{CountDownLatch, Semaphore, TimeUnit}

import scala.concurrent.duration._

import com.google.common.io.Files
import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.Path
import org.apache.hadoop.io.{BytesWritable, LongWritable, Text}
import org.apache.hadoop.mapred.TextInputFormat
import org.apache.hadoop.mapreduce.lib.input.{TextInputFormat => NewTextInputFormat}
import org.json4s.JsonAST.JArray
import org.json4s.JsonDSL._
import org.json4s.jackson.JsonMethods.{compact, render}
import org.scalatest.Matchers._
import org.scalatest.concurrent.Eventually

import org.apache.spark.ResourceName.GPU
import org.apache.spark.internal.config._
import org.apache.spark.internal.config.UI._
import org.apache.spark.scheduler.{SparkListener, SparkListenerExecutorMetricsUpdate, SparkListenerJobStart, SparkListenerTaskEnd, SparkListenerTaskStart}
import org.apache.spark.shuffle.FetchFailedException
import org.apache.spark.util.{ThreadUtils, Utils}

class SparkContextSuite extends SparkFunSuite with LocalSparkContext with Eventually {

  test("Only one SparkContext may be active at a time") {
    // Regression test for SPARK-4180
    val conf = new SparkConf().setAppName("test").setMaster("local")
    sc = new SparkContext(conf)
    val envBefore = SparkEnv.get
    // A SparkContext is already running, so we shouldn't be able to create a second one
    intercept[SparkException] { new SparkContext(conf) }
    val envAfter = SparkEnv.get
    // SparkEnv and other context variables should be the same
    assert(envBefore == envAfter)
    // After stopping the running context, we should be able to create a new one
    resetSparkContext()
    sc = new SparkContext(conf)
  }

  test("Can still construct a new SparkContext after failing to construct a previous one") {
    val conf = new SparkConf()
    // This is an invalid configuration (no app name or master URL)
    intercept[SparkException] {
      new SparkContext(conf)
    }
    // Even though those earlier calls failed, we should still be able to create a new context
    sc = new SparkContext(conf.setMaster("local").setAppName("test"))
  }

  test("Test getOrCreate") {
    var sc2: SparkContext = null
    SparkContext.clearActiveContext()
    val conf = new SparkConf().setAppName("test").setMaster("local")

    sc = SparkContext.getOrCreate(conf)

    assert(sc.getConf.get("spark.app.name").equals("test"))
    sc2 = SparkContext.getOrCreate(new SparkConf().setAppName("test2").setMaster("local"))
    assert(sc2.getConf.get("spark.app.name").equals("test"))
    assert(sc === sc2)
    assert(sc eq sc2)

    sc2.stop()
  }

  test("BytesWritable implicit conversion is correct") {
    // Regression test for SPARK-3121
    val bytesWritable = new BytesWritable()
    val inputArray = (1 to 10).map(_.toByte).toArray
    bytesWritable.set(inputArray, 0, 10)
    bytesWritable.set(inputArray, 0, 5)

    val converter = WritableConverter.bytesWritableConverter()
    val byteArray = converter.convert(bytesWritable)
    assert(byteArray.length === 5)

    bytesWritable.set(inputArray, 0, 0)
    val byteArray2 = converter.convert(bytesWritable)
    assert(byteArray2.length === 0)
  }

  test("basic case for addFile and listFiles") {
    withTempDir { dir =>
      val file1 = File.createTempFile("someprefix1", "somesuffix1", dir)
      val absolutePath1 = file1.getAbsolutePath

      val file2 = File.createTempFile("someprefix2", "somesuffix2", dir)
      val relativePath = file2.getParent + "/../" + file2.getParentFile.getName +
        "/" + file2.getName
      val absolutePath2 = file2.getAbsolutePath

      try {
        Files.write("somewords1", file1, StandardCharsets.UTF_8)
        Files.write("somewords2", file2, StandardCharsets.UTF_8)
        val length1 = file1.length()
        val length2 = file2.length()

        sc = new SparkContext(new SparkConf().setAppName("test").setMaster("local"))
        sc.addFile(file1.getAbsolutePath)
        sc.addFile(relativePath)
        sc.parallelize(Array(1), 1).map(x => {
          val gotten1 = new File(SparkFiles.get(file1.getName))
          val gotten2 = new File(SparkFiles.get(file2.getName))
          if (!gotten1.exists()) {
            throw new SparkException("file doesn't exist : " + absolutePath1)
          }
          if (!gotten2.exists()) {
            throw new SparkException("file doesn't exist : " + absolutePath2)
          }

          if (length1 != gotten1.length()) {
            throw new SparkException(
              s"file has different length $length1 than added file ${gotten1.length()} : " +
                absolutePath1)
          }
          if (length2 != gotten2.length()) {
            throw new SparkException(
              s"file has different length $length2 than added file ${gotten2.length()} : " +
                absolutePath2)
          }

          if (absolutePath1 == gotten1.getAbsolutePath) {
            throw new SparkException("file should have been copied :" + absolutePath1)
          }
          if (absolutePath2 == gotten2.getAbsolutePath) {
            throw new SparkException("file should have been copied : " + absolutePath2)
          }
          x
        }).count()
        assert(sc.listFiles().filter(_.contains("somesuffix1")).size == 1)
      } finally {
        sc.stop()
      }
    }
  }

  test("add and list jar files") {
    val jarPath = Thread.currentThread().getContextClassLoader.getResource("TestUDTF.jar")
    try {
      sc = new SparkContext(new SparkConf().setAppName("test").setMaster("local"))
      sc.addJar(jarPath.toString)
      assert(sc.listJars().filter(_.contains("TestUDTF.jar")).size == 1)
    } finally {
      sc.stop()
    }
  }

  test("SPARK-17650: malformed url's throw exceptions before bricking Executors") {
    try {
      sc = new SparkContext(new SparkConf().setAppName("test").setMaster("local"))
      Seq("http", "https", "ftp").foreach { scheme =>
        val badURL = s"$scheme://user:pwd/path"
        val e1 = intercept[MalformedURLException] {
          sc.addFile(badURL)
        }
        assert(e1.getMessage.contains(badURL))
        val e2 = intercept[MalformedURLException] {
          sc.addJar(badURL)
        }
        assert(e2.getMessage.contains(badURL))
        assert(sc.addedFiles.isEmpty)
        assert(sc.addedJars.isEmpty)
      }
    } finally {
      sc.stop()
    }
  }

  test("addFile recursive works") {
    withTempDir { pluto =>
      val neptune = Utils.createTempDir(pluto.getAbsolutePath)
      val saturn = Utils.createTempDir(neptune.getAbsolutePath)
      val alien1 = File.createTempFile("alien", "1", neptune)
      val alien2 = File.createTempFile("alien", "2", saturn)

      try {
        sc = new SparkContext(new SparkConf().setAppName("test").setMaster("local"))
        sc.addFile(neptune.getAbsolutePath, true)
        sc.parallelize(Array(1), 1).map(x => {
          val sep = File.separator
          if (!new File(SparkFiles.get(neptune.getName + sep + alien1.getName)).exists()) {
            throw new SparkException("can't access file under root added directory")
          }
          if (!new File(SparkFiles.get(
            neptune.getName + sep + saturn.getName + sep + alien2.getName)).exists()) {
            throw new SparkException("can't access file in nested directory")
          }
          if (new File(SparkFiles.get(
            pluto.getName + sep + neptune.getName + sep + alien1.getName)).exists()) {
            throw new SparkException("file exists that shouldn't")
          }
          x
        }).count()
      } finally {
        sc.stop()
      }
    }
  }

  test("addFile recursive can't add directories by default") {
    withTempDir { dir =>
      try {
        sc = new SparkContext(new SparkConf().setAppName("test").setMaster("local"))
        intercept[SparkException] {
          sc.addFile(dir.getAbsolutePath)
        }
      } finally {
        sc.stop()
      }
    }
  }

  test("cannot call addFile with different paths that have the same filename") {
    withTempDir { dir =>
      val subdir1 = new File(dir, "subdir1")
      val subdir2 = new File(dir, "subdir2")
      assert(subdir1.mkdir())
      assert(subdir2.mkdir())
      val file1 = new File(subdir1, "file")
      val file2 = new File(subdir2, "file")
      Files.write("old", file1, StandardCharsets.UTF_8)
      Files.write("new", file2, StandardCharsets.UTF_8)
      sc = new SparkContext("local-cluster[1,1,1024]", "test")
      sc.addFile(file1.getAbsolutePath)
      def getAddedFileContents(): String = {
        sc.parallelize(Seq(0)).map { _ =>
          scala.io.Source.fromFile(SparkFiles.get("file")).mkString
        }.first()
      }
      assert(getAddedFileContents() === "old")
      intercept[IllegalArgumentException] {
        sc.addFile(file2.getAbsolutePath)
      }
      assert(getAddedFileContents() === "old")
    }
  }

  // Regression tests for SPARK-16787
  for (
    schedulingMode <- Seq("local-mode", "non-local-mode");
    method <- Seq("addJar", "addFile")
  ) {
    val jarPath = Thread.currentThread().getContextClassLoader.getResource("TestUDTF.jar").toString
    val master = schedulingMode match {
      case "local-mode" => "local"
      case "non-local-mode" => "local-cluster[1,1,1024]"
    }
    test(s"$method can be called twice with same file in $schedulingMode (SPARK-16787)") {
      sc = new SparkContext(master, "test")
      method match {
        case "addJar" =>
          sc.addJar(jarPath)
          sc.addJar(jarPath)
        case "addFile" =>
          sc.addFile(jarPath)
          sc.addFile(jarPath)
      }
    }
  }

  test("add jar with invalid path") {
    withTempDir { tmpDir =>
      val tmpJar = File.createTempFile("test", ".jar", tmpDir)

      sc = new SparkContext(new SparkConf().setAppName("test").setMaster("local"))
      sc.addJar(tmpJar.getAbsolutePath)

      // Invalid jar path will only print the error log, will not add to file server.
      sc.addJar("dummy.jar")
      sc.addJar("")
      sc.addJar(tmpDir.getAbsolutePath)

      assert(sc.listJars().size == 1)
      assert(sc.listJars().head.contains(tmpJar.getName))
    }
  }

  test("SPARK-22585 addJar argument without scheme is interpreted literally without url decoding") {
    withTempDir { dir =>
      val tmpDir = new File(dir, "host%3A443")
      tmpDir.mkdirs()
      val tmpJar = File.createTempFile("t%2F", ".jar", tmpDir)

      sc = new SparkContext("local", "test")

      sc.addJar(tmpJar.getAbsolutePath)
      assert(sc.listJars().size === 1)
    }
  }

  test("Cancelling job group should not cause SparkContext to shutdown (SPARK-6414)") {
    try {
      sc = new SparkContext(new SparkConf().setAppName("test").setMaster("local"))
      val future = sc.parallelize(Seq(0)).foreachAsync(_ => {Thread.sleep(1000L)})
      sc.cancelJobGroup("nonExistGroupId")
      ThreadUtils.awaitReady(future, Duration(2, TimeUnit.SECONDS))

      // In SPARK-6414, sc.cancelJobGroup will cause NullPointerException and cause
      // SparkContext to shutdown, so the following assertion will fail.
      assert(sc.parallelize(1 to 10).count() == 10L)
    } finally {
      sc.stop()
    }
  }

  test("Comma separated paths for newAPIHadoopFile/wholeTextFiles/binaryFiles (SPARK-7155)") {
    // Regression test for SPARK-7155
    // dir1 and dir2 are used for wholeTextFiles and binaryFiles
    withTempDir { dir1 =>
      withTempDir { dir2 =>
        val dirpath1 = dir1.getAbsolutePath
        val dirpath2 = dir2.getAbsolutePath

        // file1 and file2 are placed inside dir1, they are also used for
        // textFile, hadoopFile, and newAPIHadoopFile
        // file3, file4 and file5 are placed inside dir2, they are used for
        // textFile, hadoopFile, and newAPIHadoopFile as well
        val file1 = new File(dir1, "part-00000")
        val file2 = new File(dir1, "part-00001")
        val file3 = new File(dir2, "part-00000")
        val file4 = new File(dir2, "part-00001")
        val file5 = new File(dir2, "part-00002")

        val filepath1 = file1.getAbsolutePath
        val filepath2 = file2.getAbsolutePath
        val filepath3 = file3.getAbsolutePath
        val filepath4 = file4.getAbsolutePath
        val filepath5 = file5.getAbsolutePath

        try {
          // Create 5 text files.
          Files.write("someline1 in file1\nsomeline2 in file1\nsomeline3 in file1", file1,
            StandardCharsets.UTF_8)
          Files.write("someline1 in file2\nsomeline2 in file2", file2, StandardCharsets.UTF_8)
          Files.write("someline1 in file3", file3, StandardCharsets.UTF_8)
          Files.write("someline1 in file4\nsomeline2 in file4", file4, StandardCharsets.UTF_8)
          Files.write("someline1 in file2\nsomeline2 in file5", file5, StandardCharsets.UTF_8)

          sc = new SparkContext(new SparkConf().setAppName("test").setMaster("local"))

          // Test textFile, hadoopFile, and newAPIHadoopFile for file1 and file2
          assert(sc.textFile(filepath1 + "," + filepath2).count() == 5L)
          assert(sc.hadoopFile(filepath1 + "," + filepath2,
            classOf[TextInputFormat], classOf[LongWritable], classOf[Text]).count() == 5L)
          assert(sc.newAPIHadoopFile(filepath1 + "," + filepath2,
            classOf[NewTextInputFormat], classOf[LongWritable], classOf[Text]).count() == 5L)

          // Test textFile, hadoopFile, and newAPIHadoopFile for file3, file4, and file5
          assert(sc.textFile(filepath3 + "," + filepath4 + "," + filepath5).count() == 5L)
          assert(sc.hadoopFile(filepath3 + "," + filepath4 + "," + filepath5,
            classOf[TextInputFormat], classOf[LongWritable], classOf[Text]).count() == 5L)
          assert(sc.newAPIHadoopFile(filepath3 + "," + filepath4 + "," + filepath5,
            classOf[NewTextInputFormat], classOf[LongWritable], classOf[Text]).count() == 5L)

          // Test wholeTextFiles, and binaryFiles for dir1 and dir2
          assert(sc.wholeTextFiles(dirpath1 + "," + dirpath2).count() == 5L)
          assert(sc.binaryFiles(dirpath1 + "," + dirpath2).count() == 5L)

        } finally {
          sc.stop()
        }
      }
    }
  }

  test("Default path for file based RDDs is properly set (SPARK-12517)") {
    sc = new SparkContext(new SparkConf().setAppName("test").setMaster("local"))

    // Test filetextFile, wholeTextFiles, binaryFiles, hadoopFile and
    // newAPIHadoopFile for setting the default path as the RDD name
    val mockPath = "default/path/for/"

    var targetPath = mockPath + "textFile"
    assert(sc.textFile(targetPath).name === targetPath)

    targetPath = mockPath + "wholeTextFiles"
    assert(sc.wholeTextFiles(targetPath).name === targetPath)

    targetPath = mockPath + "binaryFiles"
    assert(sc.binaryFiles(targetPath).name === targetPath)

    targetPath = mockPath + "hadoopFile"
    assert(sc.hadoopFile(targetPath).name === targetPath)

    targetPath = mockPath + "newAPIHadoopFile"
    assert(sc.newAPIHadoopFile(targetPath).name === targetPath)

    sc.stop()
  }

  test("calling multiple sc.stop() must not throw any exception") {
    noException should be thrownBy {
      sc = new SparkContext(new SparkConf().setAppName("test").setMaster("local"))
      val cnt = sc.parallelize(1 to 4).count()
      sc.cancelAllJobs()
      sc.stop()
      // call stop second time
      sc.stop()
    }
  }

  test("No exception when both num-executors and dynamic allocation set.") {
    noException should be thrownBy {
      sc = new SparkContext(new SparkConf().setAppName("test").setMaster("local")
        .set(DYN_ALLOCATION_ENABLED, true).set("spark.executor.instances", "6"))
      assert(sc.executorAllocationManager.isEmpty)
      assert(sc.getConf.getInt("spark.executor.instances", 0) === 6)
    }
  }

  test("localProperties are inherited by spawned threads.") {
    sc = new SparkContext(new SparkConf().setAppName("test").setMaster("local"))
    sc.setLocalProperty("testProperty", "testValue")
    var result = "unset";
    val thread = new Thread() { override def run() = {result = sc.getLocalProperty("testProperty")}}
    thread.start()
    thread.join()
    sc.stop()
    assert(result == "testValue")
  }

  test("localProperties do not cross-talk between threads.") {
    sc = new SparkContext(new SparkConf().setAppName("test").setMaster("local"))
    var result = "unset";
    val thread1 = new Thread() {
      override def run() = {sc.setLocalProperty("testProperty", "testValue")}}
    // testProperty should be unset and thus return null
    val thread2 = new Thread() {
      override def run() = {result = sc.getLocalProperty("testProperty")}}
    thread1.start()
    thread1.join()
    thread2.start()
    thread2.join()
    sc.stop()
    assert(result == null)
  }

  test("log level case-insensitive and reset log level") {
    sc = new SparkContext(new SparkConf().setAppName("test").setMaster("local"))
    val originalLevel = org.apache.log4j.Logger.getRootLogger().getLevel
    try {
      sc.setLogLevel("debug")
      assert(org.apache.log4j.Logger.getRootLogger().getLevel === org.apache.log4j.Level.DEBUG)
      sc.setLogLevel("INfo")
      assert(org.apache.log4j.Logger.getRootLogger().getLevel === org.apache.log4j.Level.INFO)
    } finally {
      sc.setLogLevel(originalLevel.toString)
      assert(org.apache.log4j.Logger.getRootLogger().getLevel === originalLevel)
      sc.stop()
    }
  }

  test("register and deregister Spark listener from SparkContext") {
    sc = new SparkContext(new SparkConf().setAppName("test").setMaster("local"))
    val sparkListener1 = new SparkListener { }
    val sparkListener2 = new SparkListener { }
    sc.addSparkListener(sparkListener1)
    sc.addSparkListener(sparkListener2)
    assert(sc.listenerBus.listeners.contains(sparkListener1))
    assert(sc.listenerBus.listeners.contains(sparkListener2))
    sc.removeSparkListener(sparkListener1)
    assert(!sc.listenerBus.listeners.contains(sparkListener1))
    assert(sc.listenerBus.listeners.contains(sparkListener2))
  }

  test("Cancelling stages/jobs with custom reasons.") {
    sc = new SparkContext(new SparkConf().setAppName("test").setMaster("local"))
    sc.setLocalProperty(SparkContext.SPARK_JOB_INTERRUPT_ON_CANCEL, "true")
    val REASON = "You shall not pass"

    for (cancelWhat <- Seq("stage", "job")) {
      // This countdown latch used to make sure stage or job canceled in listener
      val latch = new CountDownLatch(1)

      val listener = cancelWhat match {
        case "stage" =>
          new SparkListener {
            override def onTaskStart(taskStart: SparkListenerTaskStart): Unit = {
              sc.cancelStage(taskStart.stageId, REASON)
              latch.countDown()
            }
          }
        case "job" =>
          new SparkListener {
            override def onJobStart(jobStart: SparkListenerJobStart): Unit = {
              sc.cancelJob(jobStart.jobId, REASON)
              latch.countDown()
            }
          }
      }
      sc.addSparkListener(listener)

      val ex = intercept[SparkException] {
        sc.range(0, 10000L, numSlices = 10).mapPartitions { x =>
          x.synchronized {
            x.wait()
          }
          x
        }.count()
      }

      ex.getCause() match {
        case null =>
          assert(ex.getMessage().contains(REASON))
        case cause: SparkException =>
          assert(cause.getMessage().contains(REASON))
        case cause: Throwable =>
          fail("Expected the cause to be SparkException, got " + cause.toString() + " instead.")
      }

      latch.await(20, TimeUnit.SECONDS)
      eventually(timeout(20.seconds)) {
        assert(sc.statusTracker.getExecutorInfos.map(_.numRunningTasks()).sum == 0)
      }
      sc.removeSparkListener(listener)
    }
  }

  test("client mode with a k8s master url") {
    intercept[SparkException] {
      sc = new SparkContext("k8s://https://host:port", "test", new SparkConf())
    }
  }

  testCancellingTasks("that raise interrupted exception on cancel") {
    Thread.sleep(9999999)
  }

  // SPARK-20217 should not fail stage if task throws non-interrupted exception
  testCancellingTasks("that raise runtime exception on cancel") {
    try {
      Thread.sleep(9999999)
    } catch {
      case t: Throwable =>
        throw new RuntimeException("killed")
    }
  }

  // Launches one task that will block forever. Once the SparkListener detects the task has
  // started, kill and re-schedule it. The second run of the task will complete immediately.
  // If this test times out, then the first version of the task wasn't killed successfully.
  def testCancellingTasks(desc: String)(blockFn: => Unit): Unit = test(s"Killing tasks $desc") {
    sc = new SparkContext(new SparkConf().setAppName("test").setMaster("local"))

    SparkContextSuite.isTaskStarted = false
    SparkContextSuite.taskKilled = false
    SparkContextSuite.taskSucceeded = false

    val listener = new SparkListener {
      override def onTaskStart(taskStart: SparkListenerTaskStart): Unit = {
        eventually(timeout(10.seconds)) {
          assert(SparkContextSuite.isTaskStarted)
        }
        if (!SparkContextSuite.taskKilled) {
          SparkContextSuite.taskKilled = true
          sc.killTaskAttempt(taskStart.taskInfo.taskId, true, "first attempt will hang")
        }
      }
      override def onTaskEnd(taskEnd: SparkListenerTaskEnd): Unit = {
        if (taskEnd.taskInfo.attemptNumber == 1 && taskEnd.reason == Success) {
          SparkContextSuite.taskSucceeded = true
        }
      }
    }
    sc.addSparkListener(listener)
    eventually(timeout(20.seconds)) {
      sc.parallelize(1 to 1).foreach { x =>
        // first attempt will hang
        if (!SparkContextSuite.isTaskStarted) {
          SparkContextSuite.isTaskStarted = true
          blockFn
        }
        // second attempt succeeds immediately
      }
    }
    eventually(timeout(10.seconds)) {
      assert(SparkContextSuite.taskSucceeded)
    }
  }

  test("SPARK-19446: DebugFilesystem.assertNoOpenStreams should report " +
    "open streams to help debugging") {
    val fs = new DebugFilesystem()
    fs.initialize(new URI("file:///"), new Configuration())
    val file = File.createTempFile("SPARK19446", "temp")
    file.deleteOnExit()
    Files.write(Array.ofDim[Byte](1000), file)
    val path = new Path("file:///" + file.getCanonicalPath)
    val stream = fs.open(path)
    val exc = intercept[RuntimeException] {
      DebugFilesystem.assertNoOpenStreams()
    }
    assert(exc != null)
    assert(exc.getCause() != null)
    stream.close()
  }

  test("support barrier execution mode under local mode") {
    val conf = new SparkConf().setAppName("test").setMaster("local[2]")
    sc = new SparkContext(conf)
    val rdd = sc.makeRDD(Seq(1, 2, 3, 4), 2)
    val rdd2 = rdd.barrier().mapPartitions { it =>
      val context = BarrierTaskContext.get()
      // If we don't get the expected taskInfos, the job shall abort due to stage failure.
      if (context.getTaskInfos().length != 2) {
        throw new SparkException("Expected taksInfos length is 2, actual length is " +
          s"${context.getTaskInfos().length}.")
      }
      context.barrier()
      it
    }
    rdd2.collect()

    eventually(timeout(10.seconds)) {
      assert(sc.statusTracker.getExecutorInfos.map(_.numRunningTasks()).sum == 0)
    }
  }

  test("support barrier execution mode under local-cluster mode") {
    val conf = new SparkConf()
      .setMaster("local-cluster[3, 1, 1024]")
      .setAppName("test-cluster")
    sc = new SparkContext(conf)

    val rdd = sc.makeRDD(Seq(1, 2, 3, 4), 2)
    val rdd2 = rdd.barrier().mapPartitions { it =>
      val context = BarrierTaskContext.get()
      // If we don't get the expected taskInfos, the job shall abort due to stage failure.
      if (context.getTaskInfos().length != 2) {
        throw new SparkException("Expected taksInfos length is 2, actual length is " +
          s"${context.getTaskInfos().length}.")
      }
      context.barrier()
      it
    }
    rdd2.collect()

    eventually(timeout(10.seconds)) {
      assert(sc.statusTracker.getExecutorInfos.map(_.numRunningTasks()).sum == 0)
    }
  }

  test("cancel zombie tasks in a result stage when the job finishes") {
    val conf = new SparkConf()
      .setMaster("local-cluster[1,2,1024]")
      .setAppName("test-cluster")
      .set(UI_ENABLED.key, "false")
      // Disable this so that if a task is running, we can make sure the executor will always send
      // task metrics via heartbeat to driver.
      .set(EXECUTOR_HEARTBEAT_DROP_ZERO_ACCUMULATOR_UPDATES.key, "false")
      // Set a short heartbeat interval to send SparkListenerExecutorMetricsUpdate fast
      .set(EXECUTOR_HEARTBEAT_INTERVAL.key, "1s")
    sc = new SparkContext(conf)
    sc.setLocalProperty(SparkContext.SPARK_JOB_INTERRUPT_ON_CANCEL, "true")
    @volatile var runningTaskIds: Seq[Long] = null
    val listener = new SparkListener {
      override def onExecutorMetricsUpdate(
          executorMetricsUpdate: SparkListenerExecutorMetricsUpdate): Unit = {
        if (executorMetricsUpdate.execId != SparkContext.DRIVER_IDENTIFIER) {
          runningTaskIds = executorMetricsUpdate.accumUpdates.map(_._1)
        }
      }
    }
    sc.addSparkListener(listener)
    sc.range(0, 2).groupBy((x: Long) => x % 2, 2).map { case (x, _) =>
      val context = org.apache.spark.TaskContext.get()
      if (context.stageAttemptNumber == 0) {
        if (context.partitionId == 0) {
          // Make the first task in the first stage attempt fail.
          throw new FetchFailedException(SparkEnv.get.blockManager.blockManagerId, 0, 0, 0,
            new java.io.IOException("fake"))
        } else {
          // Make the second task in the first stage attempt sleep to generate a zombie task
          Thread.sleep(60000)
        }
      } else {
        // Make the second stage attempt successful.
      }
      x
    }.collect()
    sc.listenerBus.waitUntilEmpty(10000)
    // As executors will send the metrics of running tasks via heartbeat, we can use this to check
    // whether there is any running task.
    eventually(timeout(10.seconds)) {
      // Make sure runningTaskIds has been set
      assert(runningTaskIds != null)
      // Verify there is no running task.
      assert(runningTaskIds.isEmpty)
    }
  }

  test(s"Avoid setting ${CPUS_PER_TASK.key} unreasonably (SPARK-27192)") {
    val FAIL_REASON = s"has to be >= the task config: ${CPUS_PER_TASK.key}"
    Seq(
      ("local", 2, None),
      ("local[2]", 3, None),
      ("local[2, 1]", 3, None),
      ("spark://test-spark-cluster", 2, Option(1)),
      ("local-cluster[1, 1, 1000]", 2, Option(1)),
      ("yarn", 2, Option(1))
    ).foreach { case (master, cpusPerTask, executorCores) =>
      val conf = new SparkConf()
      conf.set(CPUS_PER_TASK, cpusPerTask)
      executorCores.map(executorCores => conf.set(EXECUTOR_CORES, executorCores))
      val ex = intercept[SparkException] {
        sc = new SparkContext(master, "test", conf)
      }
      assert(ex.getMessage.contains(FAIL_REASON))
      resetSparkContext()
    }
  }

  test("test gpu driver discovery under local-cluster mode") {
    withTempDir { dir =>
      val gpuFile = new File(dir, "gpuDiscoverScript")
      val scriptPath = mockDiscoveryScript(gpuFile,
        """'{"name": "gpu","addresses":["5", "6"]}'""")

      val conf = new SparkConf()
        .set(SPARK_DRIVER_RESOURCE_PREFIX + GPU +
          SPARK_RESOURCE_AMOUNT_SUFFIX, "1")
        .set(SPARK_DRIVER_RESOURCE_PREFIX + GPU +
          SPARK_RESOURCE_DISCOVERY_SCRIPT_SUFFIX, scriptPath)
        .setMaster("local-cluster[1, 1, 1024]")
        .setAppName("test-cluster")
      sc = new SparkContext(conf)

      // Ensure all executors has started
      eventually(timeout(10.seconds)) {
        assert(sc.statusTracker.getExecutorInfos.size == 1)
      }
      assert(sc.resources.size === 1)
      assert(sc.resources.get(GPU).get.addresses === Array("5", "6"))
      assert(sc.resources.get(GPU).get.name === "gpu")
    }
  }

  private def writeJsonFile(dir: File, strToWrite: JArray): String = {
    val f1 = File.createTempFile("test-resource-parser1", "", dir)
    JavaFiles.write(f1.toPath(), compact(render(strToWrite)).getBytes())
    f1.getPath()
  }

  test("test gpu driver resource files and discovery under local-cluster mode") {
    withTempDir { dir =>
      val gpuFile = new File(dir, "gpuDiscoverScript")
      val scriptPath = mockDiscoveryScript(gpuFile,
        """'{"name": "gpu","addresses":["5", "6"]}'""")

      val gpusAllocated =
        ("name" -> "gpu") ~
        ("addresses" -> Seq("0", "1", "8"))
      val ja = JArray(List(gpusAllocated))
      val resourcesFile = writeJsonFile(dir, ja)

      val conf = new SparkConf()
        .set(SPARK_DRIVER_RESOURCE_PREFIX + GPU +
          SPARK_RESOURCE_AMOUNT_SUFFIX, "1")
        .set(SPARK_DRIVER_RESOURCE_PREFIX + GPU +
          SPARK_RESOURCE_DISCOVERY_SCRIPT_SUFFIX, scriptPath)
        .set(DRIVER_RESOURCES_FILE, resourcesFile)
        .setMaster("local-cluster[1, 1, 1024]")
        .setAppName("test-cluster")
      sc = new SparkContext(conf)

      // Ensure all executors has started
      eventually(timeout(10.seconds)) {
        assert(sc.statusTracker.getExecutorInfos.size == 1)
      }
      // driver gpu addresses config should take precedence over the script
      assert(sc.resources.size === 1)
      assert(sc.resources.get(GPU).get.addresses === Array("0", "1", "8"))
      assert(sc.resources.get(GPU).get.name === "gpu")
    }
  }

  test("Test parsing resources task configs with missing executor config") {
    val conf = new SparkConf()
      .set(SPARK_TASK_RESOURCE_PREFIX + GPU +
        SPARK_RESOURCE_AMOUNT_SUFFIX, "1")
      .setMaster("local-cluster[1, 1, 1024]")
      .setAppName("test-cluster")

    var error = intercept[SparkException] {
      sc = new SparkContext(conf)
    }.getMessage()

    assert(error.contains("The executor resource config: spark.executor.resource.gpu.amount " +
      "needs to be specified since a task requirement config: spark.task.resource.gpu.amount " +
      "was specified"))
  }

  test("Test parsing resources executor config < task requirements") {
    val conf = new SparkConf()
      .set(SPARK_TASK_RESOURCE_PREFIX + GPU +
        SPARK_RESOURCE_AMOUNT_SUFFIX, "2")
      .set(SPARK_EXECUTOR_RESOURCE_PREFIX + GPU +
        SPARK_RESOURCE_AMOUNT_SUFFIX, "1")
      .setMaster("local-cluster[1, 1, 1024]")
      .setAppName("test-cluster")

    var error = intercept[SparkException] {
      sc = new SparkContext(conf)
    }.getMessage()

    assert(error.contains("The executor resource config: " +
      "spark.executor.resource.gpu.amount = 1 has to be >= the task config: " +
      "spark.task.resource.gpu.amount = 2"))
  }

  test("Parse resources executor config not the same multiple numbers of the task requirements") {
    val conf = new SparkConf()
      .set(SPARK_TASK_RESOURCE_PREFIX + GPU + SPARK_RESOURCE_AMOUNT_SUFFIX, "2")
      .set(SPARK_EXECUTOR_RESOURCE_PREFIX + GPU + SPARK_RESOURCE_AMOUNT_SUFFIX, "4")
      .setMaster("local-cluster[1, 1, 1024]")
      .setAppName("test-cluster")

    var error = intercept[SparkException] {
      sc = new SparkContext(conf)
    }.getMessage()

    assert(error.contains("The configuration of resource: gpu (exec = 4, task = 2) will result " +
      "in wasted resources due to resource CPU limiting the number of runnable tasks per " +
      "executor to: 1. Please adjust your configuration."))
  }

  def mockDiscoveryScript(file: File, result: String): String = {
    Files.write(s"echo $result", file, StandardCharsets.UTF_8)
    JavaFiles.setPosixFilePermissions(file.toPath(),
      EnumSet.of(OWNER_READ, OWNER_EXECUTE, OWNER_WRITE))
    file.getPath()
  }

  test("test resource scheduling under local-cluster mode") {
    import org.apache.spark.TestUtils._

    assume(!(Utils.isWindows))
    withTempDir { dir =>
      val resourceFile = new File(dir, "resourceDiscoverScript")
      val resources = """'{"name": "gpu", "addresses": ["0", "1", "2"]}'"""
      Files.write(s"echo $resources", resourceFile, StandardCharsets.UTF_8)
      JavaFiles.setPosixFilePermissions(resourceFile.toPath(),
        EnumSet.of(OWNER_READ, OWNER_EXECUTE, OWNER_WRITE))
      val discoveryScript = resourceFile.getPath()

      val conf = new SparkConf()
        .set(s"${SPARK_EXECUTOR_RESOURCE_PREFIX}${GPU}${SPARK_RESOURCE_AMOUNT_SUFFIX}", "3")
        .set(s"${SPARK_EXECUTOR_RESOURCE_PREFIX}${GPU}${SPARK_RESOURCE_DISCOVERY_SCRIPT_SUFFIX}",
          discoveryScript)
        .setMaster("local-cluster[3, 3, 1024]")
        .setAppName("test-cluster")
      setTaskResourceRequirement(conf, GPU, 1)
      sc = new SparkContext(conf)

      // Ensure all executors has started
      eventually(timeout(60.seconds)) {
        assert(sc.statusTracker.getExecutorInfos.size == 3)
      }

      val rdd = sc.makeRDD(1 to 10, 9).mapPartitions { it =>
        val context = TaskContext.get()
        context.resources().get(GPU).get.addresses.iterator
      }
      val gpus = rdd.collect()
      assert(gpus.sorted === Seq("0", "0", "0", "1", "1", "1", "2", "2", "2"))

      eventually(timeout(10.seconds)) {
        assert(sc.statusTracker.getExecutorInfos.map(_.numRunningTasks()).sum == 0)
      }
    }
  }
}

object SparkContextSuite {
  @volatile var isTaskStarted = false
  @volatile var taskKilled = false
  @volatile var taskSucceeded = false
  val semaphore = new Semaphore(0)
}


text:
module.exports =  {
	updatePlayer:function(client, player, cb) {
		var sql = 'update Player set x = ? ,y = ? , hp = ?, mp = ? , maxHp = ?, maxMp = ?, country = ?, rank = ?, level = ?, experience = ?, areaId = ?, attackValue = ?, defenceValue = ?, walkSpeed = ?, attackSpeed = ? where id = ?';
		var args = [player.x, player.y, player.hp, player.mp, player.maxHp, player.maxMp, player.country, player.rank, player.level, player.experience, player.areaId, player.attackValue, player.defenceValue, player.walkSpeed, player.attackSpeed, player.id];
		client.query(sql, args, function(err, res) {
			if(err !== null) {
				console.error('write mysql failed! ' + sql + ' ' + JSON.stringify(player) + ' stack:' + err.stack);
			}
      if(!!cb && typeof cb == 'function') {
        cb(!!err);
      }
		});
	}
};


text:
namespace KFS
{
using std::ofstream;
using std::map;
using std::vector;
using std::string;

class LayoutEmulator : public LayoutManager
{
public:
    static LayoutEmulator& Instance();
    int InitUseCurrentState(
        int64_t chunkServerTotalSpace);
    // Given a chunk->location data in a file, rebuild the chunk->location map.
    //
    int LoadChunkmap(const string& chunkLocationFn,
        bool addChunksToReplicationChecker = false);
    void AddServer(const ServerLocation& loc,
        int rack, uint64_t totalSpace, uint64_t usedSpace);
    void SetupForRebalancePlanning(
        double utilizationPercentVariationFromMean)
    {
        mVariationFromMean =
            min(1., max(0., utilizationPercentVariationFromMean * 1e-2));
    }
    int SetRebalancePlanOutFile(const string& rebalancePlanFn);
    void BuildRebalancePlan();
    bool Handle(MetaChunkReplicate& req);
    bool Handle(MetaChunkDelete& req);
    void ExecuteRebalancePlan();
    void PrintChunkserverBlockCount(ostream& os) const;
    int ReadNetworkDefn(const string& networkFn);
    int VerifyRackAwareReplication(
        bool reportAllFlag, bool verbose, ostream& os);
    seq_t  GetChunkversion(chunkId_t cid) const;
    size_t GetChunkSize(chunkId_t cid) const;
    int GetNumBlksRebalanced() const
    {
        return mNumBlksRebalanced;
    }
    void Stop()
    {
        mStopFlag = true;
    }
    int RunFsck(const string& fileName);
protected:
    LayoutEmulator()
        : mVariationFromMean(0),
          mNumBlksRebalanced(0),
          mStopFlag(false),
          mPlanFile()
    {
        SetMinChunkserversToExitRecovery(0);
        ToggleRebalancing(true);
        mReplaySetRackFlag = false;
    }
    ~LayoutEmulator();
private:
    typedef map<ServerLocation, ChunkServerPtr> Loc2Server;
    class PlacementVerifier;

    size_t RunChunkserverOps();
    void CalculateRebalaceThresholds();
    void PrepareRebalance(bool enableRebalanceFlag);
    bool Parse(const char* line, size_t size,
        bool addChunksToReplicationChecker, ServerLocation& loc);
    void ShowPlacementError(
        ostream&            os,
        const CSMap::Entry& c,
        const ChunkServer*  srv,
        string&             fileName,
        size_t              replicas,
        const char*         reason);
    void VerifyPlacement(
        const CSMap::Entry&           c,
        const Servers&                servers,
        const vector<MetaChunkInfo*>& cblk,
        ChunkPlacement&               placement,
        ostream&                      os,
        bool                          verboseFlag,
        bool                          reportAllFlag,
        PlacementVerifier&            verifier);
    size_t GetChunkSize(const CSMap::Entry& ci) const;
    void ScheduleReplication();
    void UseForPlacement(const ChunkServerPtr& srv);

    // for the purposes of rebalancing, we compute the cluster
    // wide average space utilization; then we take into the
    // desired variation from mean to compute thresholds that determine
    // which nodes are candidates for migration.
    double     mVariationFromMean;
    int        mNumBlksRebalanced;
    bool       mStopFlag;
    ofstream   mPlanFile;
private:
    // No copy.
    LayoutEmulator(const LayoutEmulator&);
    LayoutEmulator& operator=(const LayoutEmulator&);
};

}

#endif // EMULATOR_LAYOUTEMULATOR_H


text:
// -----------------------------------------------------------------------
//  <copyright file="ACLTest.cs" company="PlayFab Inc">
//    Copyright 2015 PlayFab Inc.
//
//    Licensed under the Apache License, Version 2.0 (the "License");
//    you may not use this file except in compliance with the License.
//    You may obtain a copy of the License at
//
//        http://www.apache.org/licenses/LICENSE-2.0
//
//    Unless required by applicable law or agreed to in writing, software
//    distributed under the License is distributed on an "AS IS" BASIS,
//    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
//    See the License for the specific language governing permissions and
//    limitations under the License.
//  </copyright>
// -----------------------------------------------------------------------

using Microsoft.VisualStudio.TestTools.UnitTesting;

namespace Consul.Test
{
    [TestClass]
    public class ACLTest
    {
        private const string ConsulRoot = "yep";

        [TestMethod]
        public void ACL_CreateDestroy()
        {
            if (string.IsNullOrEmpty(ConsulRoot))
            {
                Assert.Inconclusive();
            }

            var client = new Client(new ConsulClientConfiguration() {Token = ConsulRoot});
            var aclEntry = new ACLEntry()
            {
                Name = "API Test",
                Type = ACLType.Client,
                Rules = "key \"\" { policy = \"deny\" }"
            };
            var res = client.ACL.Create(aclEntry);
            var id = res.Response;

            Assert.AreNotEqual(0, res.RequestTime.TotalMilliseconds);
            Assert.IsFalse(string.IsNullOrEmpty(res.Response));

            var aclEntry2 = client.ACL.Info(id);

            Assert.IsNotNull(aclEntry2.Response);
            Assert.AreEqual(aclEntry2.Response.Name, aclEntry.Name);
            Assert.AreEqual(aclEntry2.Response.Type, aclEntry.Type);
            Assert.AreEqual(aclEntry2.Response.Rules, aclEntry.Rules);

            Assert.IsTrue(client.ACL.Destroy(id).Response);
        }

        [TestMethod]
        public void ACL_CloneUpdateDestroy()
        {
            if (string.IsNullOrEmpty(ConsulRoot))
            {
                Assert.Inconclusive();
            }

            var client = new Client(new ConsulClientConfiguration() {Token = ConsulRoot});

            var cloneRequest = client.ACL.Clone(ConsulRoot);
            var aclID = cloneRequest.Response;

            var aclEntry = client.ACL.Info(aclID);
            aclEntry.Response.Rules = "key \"\" { policy = \"deny\" }";
            client.ACL.Update(aclEntry.Response);

            var aclEntry2 = client.ACL.Info(aclID);
            Assert.AreEqual("key \"\" { policy = \"deny\" }", aclEntry2.Response.Rules);

            var id = cloneRequest.Response;

            Assert.AreNotEqual(0, cloneRequest.RequestTime.TotalMilliseconds);
            Assert.IsFalse(string.IsNullOrEmpty(aclID));

            Assert.IsTrue(client.ACL.Destroy(id).Response);
        }

        [TestMethod]
        public void ACL_Info()
        {
            if (string.IsNullOrEmpty(ConsulRoot))
            {
                Assert.Inconclusive();
            }

            var client = new Client(new ConsulClientConfiguration() {Token = ConsulRoot});

            var aclEntry = client.ACL.Info(ConsulRoot);

            Assert.IsNotNull(aclEntry.Response);
            Assert.AreNotEqual(aclEntry.RequestTime.TotalMilliseconds, 0);
            Assert.AreEqual(aclEntry.Response.ID, ConsulRoot);
            Assert.AreEqual(aclEntry.Response.Type, ACLType.Management);
        }

        [TestMethod]
        public void ACL_List()
        {
            if (string.IsNullOrEmpty(ConsulRoot))
            {
                Assert.Inconclusive();
            }
            var client = new Client(new ConsulClientConfiguration() {Token = ConsulRoot});

            var aclList = client.ACL.List();

            Assert.IsNotNull(aclList.Response);
            Assert.AreNotEqual(aclList.RequestTime.TotalMilliseconds, 0);
            Assert.IsTrue(aclList.Response.Length >= 2);
        }
    }
}


text:
A tool for creating more cohesive color schemes.

[![cohesive-colors](readme_assets/screenshot.png)](http://javierbyte.github.io/cohesive-colors/)

Thanks [Erica Schoonmaker (_erica)](https://twitter.com/_erica) for the idea.


text:
#ifndef __itktubeBasisFeatureVectorGenerator_h
#define __itktubeBasisFeatureVectorGenerator_h

#include "itktubeFeatureVectorGenerator.h"

#include <itkImage.h>

#include <vnl/vnl_matrix.h>
#include <vnl/vnl_vector.h>

#include <vector>

namespace itk
{

namespace tube
{

template< class TImage, class TLabelMap >
class BasisFeatureVectorGenerator : public FeatureVectorGenerator< TImage >
{
public:

  typedef BasisFeatureVectorGenerator            Self;
  typedef FeatureVectorGenerator< TImage >       Superclass;
  typedef SmartPointer< Self >                   Pointer;
  typedef SmartPointer< const Self >             ConstPointer;

  itkTypeMacro( BasisFeatureVectorGenerator, FeatureVectorGenerator );

  itkNewMacro( Self );

  //
  typedef typename Superclass::ImageType         ImageType;

  typedef TLabelMap                              LabelMapType;

  itkStaticConstMacro( ImageDimension, unsigned int,
    TImage::ImageDimension );

  typedef typename Superclass::IndexType         IndexType;

  typedef typename Superclass::FeatureValueType  FeatureValueType;
  typedef typename Superclass::FeatureVectorType FeatureVectorType;
  typedef typename Superclass::FeatureImageType  FeatureImageType;

  typedef FeatureVectorGenerator< TImage >       FeatureVectorGeneratorType;

  typedef typename TLabelMap::PixelType          ObjectIdType;
  typedef std::vector< ObjectIdType >            ObjectIdListType;

  typedef typename Superclass::ValueType         ValueType;
  typedef typename Superclass::ValueListType     ValueListType;

  typedef vnl_vector< ValueType >                VectorType;
  typedef std::vector< VectorType >              VectorListType;

  typedef vnl_matrix< ValueType >                MatrixType;
  typedef std::vector< MatrixType >              MatrixListType;

  void         SetInputFeatureVectorGenerator( FeatureVectorGeneratorType
                 * fGen );
  typename FeatureVectorGenerator< TImage >::Pointer
               GetInputFeatureVectorGenerator( void );

  itkSetObjectMacro( LabelMap, LabelMapType );
  itkGetObjectMacro( LabelMap, LabelMapType );

  void         SetObjectId( ObjectIdType objectId );
  void         AddObjectId( ObjectIdType objectId );
  ObjectIdType GetObjectId( unsigned int num = 0 ) const;
  unsigned     int GetNumberOfObjectIds( void ) const;

  ValueType    GetObjectMean( ObjectIdType objectId ) const;
  void         SetObjectMean( ObjectIdType objectId, ValueType val );
  MatrixType   GetObjectCovariance( ObjectIdType objectId ) const;
  void         SetObjectCovariance( ObjectIdType objectId, MatrixType val );

  ValueType    GetGlobalMean( void ) const;
  void         SetGlobalMean( ValueType val );
  MatrixType   GetGlobalCovariance( void ) const;
  void         SetGlobalCovariance( MatrixType val );

  unsigned     int    GetNumberOfBasis( void ) const;

  double       GetBasisValue( unsigned int basisNum ) const;
  VectorType   GetBasisVector( unsigned int basisNum ) const;
  MatrixType   GetBasisMatrix( void ) const;
  VectorType   GetBasisValues( void ) const;

  void         SetBasisValue( unsigned int basisNum, double value );
  void         SetBasisVector( unsigned int basisNum, const VectorType & vec );
  void         SetBasisMatrix( const MatrixType & mat );
  void         SetBasisValues( const VectorType & values );

  typename FeatureImageType::Pointer GetFeatureImage( unsigned int fNum ) const;

  virtual void GenerateBasis( void );

  void SetNumberOfBasisToUseAsFeatures( unsigned int numBasisUsed );
  void SetNumberOfLDABasisToUseAsFeatures( unsigned int numBasisUsed );

  virtual unsigned int      GetNumberOfFeatures( void ) const;

  virtual FeatureVectorType GetFeatureVector( const IndexType & indx ) const;

  virtual FeatureValueType  GetFeatureVectorValue( const IndexType & indx,
    unsigned int fNum ) const;

protected:

  BasisFeatureVectorGenerator( void );
  virtual ~BasisFeatureVectorGenerator( void );

  void PrintSelf( std::ostream & os, Indent indent ) const;

private:

  // Purposely not implemented
  BasisFeatureVectorGenerator( const Self & );
  void operator = ( const Self & );      // Purposely not implemented

  //  Data
  typename FeatureVectorGeneratorType::Pointer m_InputFeatureVectorGenerator;

  typename LabelMapType::Pointer  m_LabelMap;

  ObjectIdListType                m_ObjectIdList;
  VectorListType                  m_ObjectMeanList;
  MatrixListType                  m_ObjectCovarianceList;

  VectorType                      m_GlobalMean;
  MatrixType                      m_GlobalCovariance;

  unsigned int                    m_NumberOfBasis;
  unsigned int                    m_NumberOfBasisToUseAsFeatures;
  unsigned int                    m_NumberOfLDABasisToUseAsFeatures;

  MatrixType                      m_BasisMatrix;
  VectorType                      m_BasisValues;
}; // End class BasisFeatureVectorGenerator

} // End namespace tube

} // End namespace itk

#ifndef ITK_MANUAL_INSTANTIATION
#include "itktubeBasisFeatureVectorGenerator.hxx"
#endif

#endif // End !defined(__itktubeBasisFeatureVectorGenerator_h)


text:
#include <stdio.h>
#include <stdlib.h>

#include <ida/ida_spgmr.h>
#include "ida_spils_impl.h"
#include "ida_impl.h"

#include <sundials/sundials_spgmr.h>
#include <sundials/sundials_math.h>

/* Constants */

#define ZERO         RCONST(0.0)
#define ONE          RCONST(1.0)
#define PT9          RCONST(0.9)
#define PT05         RCONST(0.05)

/* IDASPGMR linit, lsetup, lsolve, lperf, and lfree routines */

static int IDASpgmrInit(IDAMem IDA_mem);

static int IDASpgmrSetup(IDAMem IDA_mem,
                         N_Vector yy_p, N_Vector yp_p, N_Vector rr_p,
                         N_Vector tmp1, N_Vector tmp2, N_Vector tmp3);

static int IDASpgmrSolve(IDAMem IDA_mem, N_Vector bb, N_Vector weight,
                         N_Vector yy_now, N_Vector yp_now, N_Vector rr_now);

static int IDASpgmrPerf(IDAMem IDA_mem, int perftask);

static int IDASpgmrFree(IDAMem IDA_mem);

/* Readability Replacements */

#define nst          (IDA_mem->ida_nst)
#define tn           (IDA_mem->ida_tn)
#define cj           (IDA_mem->ida_cj)
#define epsNewt      (IDA_mem->ida_epsNewt)
#define res          (IDA_mem->ida_res)
#define user_data    (IDA_mem->ida_user_data)
#define ewt          (IDA_mem->ida_ewt)
#define errfp        (IDA_mem->ida_errfp)
#define linit        (IDA_mem->ida_linit)
#define lsetup       (IDA_mem->ida_lsetup)
#define lsolve       (IDA_mem->ida_lsolve)
#define lperf        (IDA_mem->ida_lperf)
#define lfree        (IDA_mem->ida_lfree)
#define lmem         (IDA_mem->ida_lmem)
#define nni          (IDA_mem->ida_nni)
#define ncfn         (IDA_mem->ida_ncfn)
#define setupNonNull (IDA_mem->ida_setupNonNull)
#define vec_tmpl     (IDA_mem->ida_tempv1)

#define sqrtN     (idaspils_mem->s_sqrtN)
#define epslin    (idaspils_mem->s_epslin)
#define ytemp     (idaspils_mem->s_ytemp)
#define yptemp    (idaspils_mem->s_yptemp)
#define xx        (idaspils_mem->s_xx)
#define ycur      (idaspils_mem->s_ycur)
#define ypcur     (idaspils_mem->s_ypcur)
#define rcur      (idaspils_mem->s_rcur)
#define npe       (idaspils_mem->s_npe)
#define nli       (idaspils_mem->s_nli)
#define nps       (idaspils_mem->s_nps)
#define ncfl      (idaspils_mem->s_ncfl)
#define nst0      (idaspils_mem->s_nst0)
#define nni0      (idaspils_mem->s_nni0)
#define nli0      (idaspils_mem->s_nli0)
#define ncfn0     (idaspils_mem->s_ncfn0)
#define ncfl0     (idaspils_mem->s_ncfl0)
#define nwarn     (idaspils_mem->s_nwarn)
#define njtimes   (idaspils_mem->s_njtimes)
#define nres      (idaspils_mem->s_nres)
#define spils_mem (idaspils_mem->s_spils_mem)

#define jtimesDQ  (idaspils_mem->s_jtimesDQ)
#define jtimes    (idaspils_mem->s_jtimes)
#define jdata     (idaspils_mem->s_jdata)

#define last_flag (idaspils_mem->s_last_flag)

/*
 * -----------------------------------------------------------------
 * IDASpgmr
 * -----------------------------------------------------------------
 *
 * This routine initializes the memory record and sets various function
 * fields specific to the IDASPGMR linear solver module.
 *
 * IDASpgmr first calls the existing lfree routine if this is not NULL.
 * It then sets the ida_linit, ida_lsetup, ida_lsolve, ida_lperf, and
 * ida_lfree fields in (*IDA_mem) to be IDASpgmrInit, IDASpgmrSetup,
 * IDASpgmrSolve, IDASpgmrPerf, and IDASpgmrFree, respectively.
 * It allocates memory for a structure of type IDASpilsMemRec and sets
 * the ida_lmem field in (*IDA_mem) to the address of this structure.
 * It sets setupNonNull in (*IDA_mem).  It then various fields in the
 * IDASpilsMemRec structure. Finally, IDASpgmr allocates memory for
 * ytemp, yptemp, and xx, and calls SpgmrMalloc to allocate memory
 * for the Spgmr solver.
 *
 * The return value of IDASpgmr is:
 *   IDASPILS_SUCCESS       = 0  if successful
 *   IDASPILS_MEM_FAIL     = -1 if IDA_mem is NULL or a memory allocation failed
 *   IDASPILS_ILL_INPUT = -2 if the gstype argument is illegal.
 *
 * -----------------------------------------------------------------
 */

int IDASpgmr(void *ida_mem, int maxl)
{
  IDAMem IDA_mem;
  IDASpilsMem idaspils_mem;
  SpgmrMem spgmr_mem;
  int flag, maxl1;

  /* Return immediately if ida_mem is NULL */
  if (ida_mem == NULL) {
    IDAProcessError(NULL, IDASPILS_MEM_NULL, "IDASPGMR", "IDASpgmr", MSGS_IDAMEM_NULL);
    return(IDASPILS_MEM_NULL);
  }
  IDA_mem = (IDAMem) ida_mem;

  /* Check if N_VDotProd is present */
  if(vec_tmpl->ops->nvdotprod == NULL) {
    IDAProcessError(NULL, IDASPILS_ILL_INPUT, "IDASPGMR", "IDASpgmr", MSGS_BAD_NVECTOR);
    return(IDASPILS_ILL_INPUT);
  }

  if (lfree != NULL) flag = lfree((IDAMem) ida_mem);

  /* Set five main function fields in ida_mem */
  linit  = IDASpgmrInit;
  lsetup = IDASpgmrSetup;
  lsolve = IDASpgmrSolve;
  lperf  = IDASpgmrPerf;
  lfree  = IDASpgmrFree;

  /* Get memory for IDASpilsMemRec */
  idaspils_mem = NULL;
  idaspils_mem = (IDASpilsMem) malloc(sizeof(struct IDASpilsMemRec));
  if (idaspils_mem == NULL) {
    IDAProcessError(NULL, IDASPILS_MEM_FAIL, "IDASPGMR", "IDASpgmr", MSGS_MEM_FAIL);
    return(IDASPILS_MEM_FAIL);
  }

  /* Set ILS type */
  idaspils_mem->s_type = SPILS_SPGMR;

  /* Set SPGMR parameters that were passed in call sequence */
  maxl1 = (maxl <= 0) ? IDA_SPILS_MAXL : maxl;
  idaspils_mem->s_maxl     = maxl1;

  /* Set defaults for Jacobian-related fileds */
  jtimesDQ = TRUE;
  jtimes   = NULL;
  jdata    = NULL;

  /* Set defaults for preconditioner-related fields */
  idaspils_mem->s_pset   = NULL;
  idaspils_mem->s_psolve = NULL;
  idaspils_mem->s_pfree  = NULL;
  idaspils_mem->s_pdata  = IDA_mem->ida_user_data;

  /* Set default values for the rest of the Spgmr parameters */
  idaspils_mem->s_gstype   = MODIFIED_GS;
  idaspils_mem->s_maxrs    = IDA_SPILS_MAXRS;
  idaspils_mem->s_eplifac  = PT05;
  idaspils_mem->s_dqincfac = ONE;

  idaspils_mem->s_last_flag  = IDASPILS_SUCCESS;

  /* Set setupNonNull to FALSE */
  setupNonNull = FALSE;

  /* Allocate memory for ytemp, yptemp, and xx */

  ytemp = N_VClone(vec_tmpl);
  if (ytemp == NULL) {
    IDAProcessError(NULL, IDASPILS_MEM_FAIL, "IDASPGMR", "IDASpgmr", MSGS_MEM_FAIL);
    free(idaspils_mem); idaspils_mem = NULL;
    return(IDASPILS_MEM_FAIL);
  }

  yptemp = N_VClone(vec_tmpl);
  if (yptemp == NULL) {
    IDAProcessError(NULL, IDASPILS_MEM_FAIL, "IDASPGMR", "IDASpgmr", MSGS_MEM_FAIL);
    N_VDestroy(ytemp);
    free(idaspils_mem); idaspils_mem = NULL;
    return(IDASPILS_MEM_FAIL);
  }

  xx = N_VClone(vec_tmpl);
  if (xx == NULL) {
    IDAProcessError(NULL, IDASPILS_MEM_FAIL, "IDASPGMR", "IDASpgmr", MSGS_MEM_FAIL);
    N_VDestroy(ytemp);
    N_VDestroy(yptemp);
    free(idaspils_mem); idaspils_mem = NULL;
    return(IDASPILS_MEM_FAIL);
  }

  /* Compute sqrtN from a dot product */
  N_VConst(ONE, ytemp);
  sqrtN = SUNRsqrt( N_VDotProd(ytemp, ytemp) );

  /* Call SpgmrMalloc to allocate workspace for Spgmr */
  spgmr_mem = NULL;
  spgmr_mem = SpgmrMalloc(maxl1, vec_tmpl);
  if (spgmr_mem == NULL) {
    IDAProcessError(NULL, IDASPILS_MEM_FAIL, "IDASPGMR", "IDASpgmr", MSGS_MEM_FAIL);
    N_VDestroy(ytemp);
    N_VDestroy(yptemp);
    N_VDestroy(xx);
    free(idaspils_mem); idaspils_mem = NULL;
    return(IDASPILS_MEM_FAIL);
  }

  /* Attach SPGMR memory to spils memory structure */
  spils_mem = (void *)spgmr_mem;

  /* Attach linear solver memory to the integrator memory */
  lmem = idaspils_mem;

  return(IDASPILS_SUCCESS);
}

/*
 * -----------------------------------------------------------------
 * IDASPGMR interface routines
 * -----------------------------------------------------------------
 */

/* Additional readability Replacements */

#define gstype   (idaspils_mem->s_gstype)
#define maxl     (idaspils_mem->s_maxl)
#define maxrs    (idaspils_mem->s_maxrs)
#define eplifac  (idaspils_mem->s_eplifac)
#define psolve   (idaspils_mem->s_psolve)
#define pset     (idaspils_mem->s_pset)
#define pdata    (idaspils_mem->s_pdata)

static int IDASpgmrInit(IDAMem IDA_mem)
{
  IDASpilsMem idaspils_mem;

  idaspils_mem = (IDASpilsMem) lmem;

  /* Initialize counters */
  npe = nli = nps = ncfl = 0;
  njtimes = nres = 0;

  /* Set setupNonNull to TRUE iff there is preconditioning with setup */
  setupNonNull = (psolve != NULL) && (pset != NULL);

  /* Set Jacobian-related fields, based on jtimesDQ */
  if (jtimesDQ) {
    jtimes = IDASpilsDQJtimes;
    jdata = IDA_mem;
  } else {
    jdata = user_data;
  }

  last_flag = IDASPILS_SUCCESS;
  return(0);
}

static int IDASpgmrSetup(IDAMem IDA_mem,
                         N_Vector yy_p, N_Vector yp_p, N_Vector rr_p,
                         N_Vector tmp1, N_Vector tmp2, N_Vector tmp3)
{
  int retval;
  IDASpilsMem idaspils_mem;

  idaspils_mem = (IDASpilsMem) lmem;

  /* Call user setup routine pset and update counter npe. */
  retval = pset(tn, yy_p, yp_p, rr_p, cj, pdata,
                tmp1, tmp2, tmp3);
  npe++;

  /* Return flag showing success or failure of pset. */
  if (retval < 0) {
    IDAProcessError(IDA_mem, SPGMR_PSET_FAIL_UNREC, "IDASPGMR", "IDASpgmrSetup", MSGS_PSET_FAILED);
    last_flag = SPGMR_PSET_FAIL_UNREC;
    return(-1);
  }
  if (retval > 0) {
    last_flag = SPGMR_PSET_FAIL_REC;
    return(+1);
  }

  last_flag = SPGMR_SUCCESS;
  return(0);
}

/*
 * The x-scaling and b-scaling arrays are both equal to weight.
 *
 * We set the initial guess, x = 0, then call SpgmrSolve.
 * We copy the solution x into b, and update the counters nli, nps, ncfl.
 * If SpgmrSolve returned nli_inc = 0 (hence x = 0), we take the SPGMR
 * vtemp vector (= P_inverse F) as the correction vector instead.
 *  Finally, we set the return value according to the success of SpgmrSolve.
 */

static int IDASpgmrSolve(IDAMem IDA_mem, N_Vector bb, N_Vector weight,
                         N_Vector yy_now, N_Vector yp_now, N_Vector rr_now)
{
  IDASpilsMem idaspils_mem;
  SpgmrMem spgmr_mem;
  int pretype, nli_inc, nps_inc, retval;
  realtype res_norm;

  idaspils_mem = (IDASpilsMem) lmem;

  spgmr_mem = (SpgmrMem) spils_mem;

  /* Set SpgmrSolve convergence test constant epslin, in terms of the
    Newton convergence test constant epsNewt and safety factors.  The factor
    sqrt(Neq) assures that the GMRES convergence test is applied to the
    WRMS norm of the residual vector, rather than the weighted L2 norm. */
  epslin = sqrtN*eplifac*epsNewt;

  /* Set vectors ycur, ypcur, and rcur for use by the Atimes and Psolve */
  ycur = yy_now;
  ypcur = yp_now;
  rcur = rr_now;

  /* Set SpgmrSolve inputs pretype and initial guess xx = 0. */
  pretype = (psolve == NULL) ? PREC_NONE : PREC_LEFT;
  N_VConst(ZERO, xx);

  /* Call SpgmrSolve and copy xx to bb. */
  retval = SpgmrSolve(spgmr_mem, IDA_mem, xx, bb, pretype, gstype, epslin,
                      maxrs, IDA_mem, weight, weight, IDASpilsAtimes,
                      IDASpilsPSolve, &res_norm, &nli_inc, &nps_inc);

  if (nli_inc == 0) N_VScale(ONE, SPGMR_VTEMP(spgmr_mem), bb);
  else N_VScale(ONE, xx, bb);

  /* Increment counters nli, nps, and return if successful. */
  nli += nli_inc;
  nps += nps_inc;
  if (retval != SPGMR_SUCCESS) ncfl++;

  /* Interpret return value from SpgmrSolve */

  last_flag = retval;

  switch(retval) {

  case SPGMR_SUCCESS:
    return(0);
    break;
  case SPGMR_RES_REDUCED:
    return(1);
    break;
  case SPGMR_CONV_FAIL:
    return(1);
    break;
  case SPGMR_QRFACT_FAIL:
    return(1);
    break;
  case SPGMR_PSOLVE_FAIL_REC:
    return(1);
    break;
  case SPGMR_ATIMES_FAIL_REC:
    return(1);
    break;
  case SPGMR_MEM_NULL:
    return(-1);
    break;
  case SPGMR_ATIMES_FAIL_UNREC:
    IDAProcessError(IDA_mem, SPGMR_ATIMES_FAIL_UNREC, "IDASPGMR", "IDASpgmrSolve", MSGS_JTIMES_FAILED);
    return(-1);
    break;
  case SPGMR_PSOLVE_FAIL_UNREC:
    IDAProcessError(IDA_mem, SPGMR_PSOLVE_FAIL_UNREC, "IDASPGMR", "IDASpgmrSolve", MSGS_PSOLVE_FAILED);
    return(-1);
    break;
  case SPGMR_GS_FAIL:
    return(-1);
    break;
  case SPGMR_QRSOL_FAIL:
    return(-1);
    break;
  }

  return(0);
}

/*
 * This routine handles performance monitoring specific to the IDASPGMR
 * linear solver.  When perftask = 0, it saves values of various counters.
 * When perftask = 1, it examines difference quotients in these counters,
 * and depending on their values, it prints up to three warning messages.
 * Messages are printed up to a maximum of 10 times.
 */

static int IDASpgmrPerf(IDAMem IDA_mem, int perftask)
{
  IDASpilsMem idaspils_mem;
  realtype avdim, rcfn, rcfl;
  long int nstd, nnid;
  booleantype lavd, lcfn, lcfl;

  idaspils_mem = (IDASpilsMem) lmem;

  if (perftask == 0) {
    nst0 = nst;  nni0 = nni;  nli0 = nli;
    ncfn0 = ncfn;  ncfl0 = ncfl;
    nwarn = 0;
    return(0);
  }

  nstd = nst - nst0;  nnid = nni - nni0;
  if (nstd == 0 || nnid == 0) return(0);
  avdim = (realtype) ((nli - nli0)/((realtype) nnid));
  rcfn = (realtype) ((ncfn - ncfn0)/((realtype) nstd));
  rcfl = (realtype) ((ncfl - ncfl0)/((realtype) nnid));
  lavd = (avdim > ((realtype) maxl ));
  lcfn = (rcfn > PT9);
  lcfl = (rcfl > PT9);
  if (!(lavd || lcfn || lcfl)) return(0);
  nwarn++;
  if (nwarn > 10) return(1);
  if (lavd)
    IDAProcessError(IDA_mem, IDA_WARNING, "IDASPGMR", "IDASpgmrPerf", MSGS_AVD_WARN, tn, avdim);
  if (lcfn)
    IDAProcessError(IDA_mem, IDA_WARNING, "IDASPGMR", "IDASpgmrPerf", MSGS_CFN_WARN, tn, rcfn);
  if (lcfl)
    IDAProcessError(IDA_mem, IDA_WARNING, "IDASPGMR", "IDASpgmrPerf", MSGS_CFL_WARN, tn, rcfl);

  return(0);
}

static int IDASpgmrFree(IDAMem IDA_mem)
{
  IDASpilsMem idaspils_mem;
  SpgmrMem spgmr_mem;

  idaspils_mem = (IDASpilsMem) lmem;

  N_VDestroy(ytemp);
  N_VDestroy(yptemp);
  N_VDestroy(xx);

  spgmr_mem = (SpgmrMem) spils_mem;
  SpgmrFree(spgmr_mem);

  if (idaspils_mem->s_pfree != NULL) (idaspils_mem->s_pfree)(IDA_mem);

  free(idaspils_mem); idaspils_mem = NULL;

  return(0);
}


text:
//
//  ELThumbnail.c
//  ELMediaPlayerLib
//
//  Created by www.e-linkway.com on 12-6-25.
//  Copyright (c) 2012年 www.e-linkway.com. All rights reserved.
//

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <math.h>

#include "libavformat/avformat.h"
#include "libswscale/swscale.h"
#include "libavcodec/avcodec.h"

#include "ELThumbnail.h"

#define MAX_THUMB_DIMENSION 256         /* max dimension in pixels */
#define MAX_THUMB_BYTES (100*1024)

@implementation ELThumbnail

@synthesize videoPath = _videoPath;

-(id) initWithVideoPath: (NSString *) videoPath
{
    self = [super init];

    if ( !videoPath )
    {
        [self release];

        return nil;
    }

    self.videoPath = videoPath;

    return self;
}

-(void)dealloc
{
    self.videoPath = nil;

    [super dealloc];
}

-(NSData *) pngData
{
    int ret = [self thumbnail_video_read];

    if (0 != ret)
    {
        return nil;
    }

    if (!_imageData)
    {
        return nil;
    }

    // 在parseFrame中,为_imageData赋予了值
//    NSData *data = [NSData dataWithBytesNoCopy: _imageData length: _dataSize freeWhenDone: YES];

    NSData *data = [NSData dataWithBytes: _imageData length: _dataSize];

    av_free(_imageData);

//    [data writeToFile: [NSString stringWithFormat: @"/tmp/%d.png", _keyFrameIndex] atomically:NO];

    return data;
}

////////////////////////////////////////////////////////////////////////////////////////////////////

static void thumbnail_calculate_dimensions(int src_width, int src_height,
                                           int src_sar_num, int src_sar_den,
                                           int *dst_width, int *dst_height);

static size_t thumbnail_create(int src_width, int src_height, int src_stride[],
                               enum PixelFormat src_pixfmt, const uint8_t * const src_data[],
                               int dst_width, int dst_height,
                               uint8_t **output_data, size_t output_max_size);

////////////////////////////////////////////////////////////////////////////////////////////////////

//
/* open a given stream. Return 0 if OK */
-(int) thumbnail_video_stream_component_open
{
    _codecContext = _formatContext->streams[_videoStreamIndex]->codec;
    _videoCodec = avcodec_find_decoder(  _codecContext->codec_id );

    if (!_videoCodec)
    {
        return -1;
    }

    _codecContext->workaround_bugs   = FF_BUG_AUTODETECT;     // auto detection
    _codecContext->lowres            = 0;  // 缩放最小比例
    if(_codecContext->lowres > _videoCodec->max_lowres)       // 支持的最小缩放比例: max_lowres
    {
        av_log(_codecContext, AV_LOG_WARNING, "The maximum value for lowres supported by the decoder is %d\n",
               _videoCodec->max_lowres);

        _codecContext->lowres= _videoCodec->max_lowres;
    }

    _codecContext->idct_algo         = FF_IDCT_AUTO;
    _codecContext->skip_frame        = AVDISCARD_DEFAULT;
    _codecContext->skip_idct         = AVDISCARD_DEFAULT;
    _codecContext->skip_loop_filter  = AVDISCARD_DEFAULT;
    _codecContext->error_concealment = FF_EC_DEBLOCK;

    if(_codecContext->lowres)
    {
        _codecContext->flags |= CODEC_FLAG_EMU_EDGE;
    }

    if(_videoCodec->capabilities & CODEC_CAP_DR1)
    {
        _codecContext->flags |= CODEC_FLAG_EMU_EDGE;
    }

    if (    !_videoCodec
        ||  avcodec_open2(_codecContext, _videoCodec, NULL) < 0)
    {
        return -1;
    }

    _formatContext->streams[_videoStreamIndex]->discard = AVDISCARD_DEFAULT;

    return 0;
}

-(int) parseFrame: (AVFrame *) pAVFrame
{
    int thumb_width = 0;
    int thumb_height = 0;

    thumbnail_calculate_dimensions (_codecContext->width, _codecContext->height,
                                    _codecContext->sample_aspect_ratio.num,
                                    _codecContext->sample_aspect_ratio.den,
                                    &thumb_width, &thumb_height);

    int size = thumbnail_create (_codecContext->width,
                                 _codecContext->height,
                                 pAVFrame->linesize,
                                 _codecContext->pix_fmt,
                                 (const uint8_t * const*) pAVFrame->data,
                                 thumb_width,
                                 thumb_height,
                                 &_imageData,
                                 MAX_THUMB_BYTES);

    if( size > 0 )
    {
        _dataSize = size;

        return 0;
    }
    else
    {
        return -1;
    }
}

-(int) thumbnail_video_read
{
    int ret = -1;

    AVPacket avPacket, *pAVPacket = &avPacket;

    _formatContext = NULL;

    int _offsetSeconds = 5;

    int err = avformat_open_input(&_formatContext, [_videoPath UTF8String], NULL, NULL);
    if (err < 0)
    {
        goto open_failed;
    }

    err = avformat_find_stream_info(_formatContext, NULL);
    if (err < 0)
    {
        goto find_stream_info_failed;
    }

    int nb_streams = _formatContext->nb_streams;

    if (_formatContext->pb)
    {
        _formatContext->pb->eof_reached = 0; // FIXME hack, ffplay maybe should not use url_feof() to test for the end
    }

    for (int i = 0; i < nb_streams; i++)
    {
        _formatContext->streams[i]->discard = AVDISCARD_ALL;
    }

    _videoStreamIndex = av_find_best_stream(_formatContext, AVMEDIA_TYPE_VIDEO, -1, -1, NULL, 0);

    if (_videoStreamIndex >= 0)
    {
        if ([self thumbnail_video_stream_component_open] < 0)
        {
            goto video_stream_component_open_failed;
        }
    }

    AVFrame *videoFrame = avcodec_alloc_frame();

    // TODO: Seek到某个时间
    // 查找这个时间附近的关键帧

    int iCurrentKeyFrameCount = 0;
    // 读数据, 直到读完一帧
    for (;;)
    {
        if ( 0 > av_read_frame( _formatContext, pAVPacket) )
        {
            // 读到文件末尾了
            if (    ret == AVERROR_EOF
                ||  url_feof( _formatContext->pb))
            {
                goto read_key_frame_failed;
            }

            // 出错了
            if (    _formatContext->pb
                &&  _formatContext->pb->error)
            {
                goto read_key_frame_failed;
            }

            continue;
        }

        int got_picture = 0;

        avcodec_decode_video2(_codecContext, videoFrame, &got_picture, pAVPacket);

        if (!got_picture)
        {
            av_free_packet(pAVPacket);

            continue;
        }

        if(  iCurrentKeyFrameCount < _offsetSeconds )
        {
            iCurrentKeyFrameCount++;

            continue;
        }

        // 上面都OK了,获取thumbnail
        // thumbnail放到了成员变量中,明天继续调试
        ret = [self parseFrame: videoFrame];

        break;
    }

read_key_frame_failed:

    avcodec_free_frame(&videoFrame);    // 这里会不会有内存泄露?

video_stream_component_open_failed:

    // 关闭流
    if (_videoStreamIndex >= 0)
    {
        _formatContext->streams[_videoStreamIndex]->discard = AVDISCARD_ALL;
        avcodec_close(_codecContext);
    }

find_stream_info_failed:
    avformat_free_context(_formatContext);
    _formatContext = NULL;

open_failed:

    return ret;
}

@end

#pragma mark -
/////////////////////////////////////////////////////////////////////////////////////////
/* calculate the thumbnail dimensions, taking pixel aspect into account */
static void thumbnail_calculate_dimensions(int src_width,
                                           int src_height,
                                           int src_sar_num,
                                           int src_sar_den,
                                           int *dst_width,
                                           int *dst_height)
{
    if (src_sar_num <= 0 || src_sar_den <= 0)
    {
        src_sar_num = 1;
        src_sar_den = 1;
    }
    if ((src_width * src_sar_num) / src_sar_den > src_height)
    {
        *dst_width = MAX_THUMB_DIMENSION;
        *dst_height = (*dst_width * src_height) /
        ((src_width * src_sar_num) / src_sar_den);
    }
    else
    {
        *dst_height = MAX_THUMB_DIMENSION;
        *dst_width = (*dst_height *
                      ((src_width * src_sar_num) / src_sar_den)) /
        src_height;
    }

    if (*dst_width < 8)
    {
        *dst_width = 8;
    }

    if (*dst_height < 1)
    {
        *dst_height = 1;
    }
}

/*
 * Rescale and encode a PNG thumbnail
 * on success, fills in output_data and returns the number of bytes used
 */
static size_t thumbnail_create(
                               int src_width,
                               int src_height,
                               int src_stride[],
                               enum PixelFormat src_pixfmt,
                               const uint8_t * const src_data[],
                               int dst_width,
                               int dst_height,
                               uint8_t **output_data,
                               size_t output_max_size)
{
    AVCodecContext *encoder_codec_ctx = NULL;
    AVCodec *encoder_codec = NULL;
    struct SwsContext *scaler_ctx = NULL;
    int sws_flags = SWS_BILINEAR;
    AVFrame *dst_frame = NULL;
    uint8_t *dst_buffer = NULL;
    uint8_t *encoder_output_buffer = NULL;
    size_t encoder_output_buffer_size;
    int err;

    // 2012.10.11 在SDK init的时候,执行av_register命令
    av_register_all();

    // 寻找png编码器
    encoder_codec = avcodec_find_encoder_by_name ("png");
    if (encoder_codec == NULL)
    {
        return 0;
    }

    /* NOTE: the scaler will be used even if the src and dst image dimensions
     * match, because the scaler will also perform colour space conversion */
    scaler_ctx = sws_getContext (src_width, src_height, src_pixfmt,
                                 dst_width, dst_height, PIX_FMT_RGB24,
                                 sws_flags, NULL, NULL, NULL);

    if (scaler_ctx == NULL)
    {
        return 0;
    }

    dst_frame = avcodec_alloc_frame ();
    if (dst_frame == NULL)
    {
        sws_freeContext(scaler_ctx);

        return 0;
    }

    dst_buffer = av_malloc (avpicture_get_size (PIX_FMT_RGB24, dst_width, dst_height));
    if (dst_buffer == NULL)
    {
        av_free (dst_frame);
        sws_freeContext(scaler_ctx);

        return 0;
    }

    avpicture_fill ((AVPicture *) dst_frame, dst_buffer,
                    PIX_FMT_RGB24, dst_width, dst_height);

    sws_scale (scaler_ctx,
               src_data,
               src_stride,
               0, src_height,
               dst_frame->data,
               dst_frame->linesize);

    encoder_output_buffer_size = output_max_size;
    encoder_output_buffer = av_malloc (encoder_output_buffer_size);
    if (encoder_output_buffer == NULL)
    {
        av_free (dst_buffer);
        av_free (dst_frame);
        sws_freeContext(scaler_ctx);

        return 0;
    }

    encoder_codec_ctx = avcodec_alloc_context ();
    if (encoder_codec_ctx == NULL)
    {
        av_free (encoder_output_buffer);
        av_free (dst_buffer);
        av_free (dst_frame);
        sws_freeContext(scaler_ctx);

        return 0;
    }

    encoder_codec_ctx->width = dst_width;
    encoder_codec_ctx->height = dst_height;
    encoder_codec_ctx->pix_fmt = PIX_FMT_RGB24;

    if (avcodec_open (encoder_codec_ctx, encoder_codec) < 0)
    {
        av_free (encoder_codec_ctx);
        av_free (encoder_output_buffer);
        av_free (dst_buffer);
        av_free (dst_frame);
        sws_freeContext(scaler_ctx);

        return 0;
    }

    err = avcodec_encode_video (encoder_codec_ctx,
                                encoder_output_buffer,
                                encoder_output_buffer_size, dst_frame);

    avcodec_close (encoder_codec_ctx);
    av_free (encoder_codec_ctx);
    av_free (dst_buffer);
    av_free (dst_frame);
    sws_freeContext(scaler_ctx);

    *output_data = encoder_output_buffer;

    return err < 0 ? 0 : err;
}