/*
 * LOGICAL-PARADOX.ORG
 * Copyright (C)2005 satoshi akabane(akabane@logical-paradox.org)
 * $Id: DomainConnection.java,v 1.2 2005/09/07 13:38:06 rampil Exp $
 */
package org.logical_paradox.vsan.dfs;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.logical_paradox.vsan.dac.DacConnection;
import org.logical_paradox.vsan.dfs.query.DacBestStoreSpaceQueryCommand;
import org.logical_paradox.vsan.dfs.query.DacQueryCommand;
import org.logical_paradox.vsan.dfs.query.DacQueryThread;
import org.logical_paradox.vsan.domain.Domain;
import org.logical_paradox.vsan.domain.exception.DomainAccessException;
import org.logical_paradox.vsan.node.DiskArrayNode;

/**
 * ドメインに対してアクセスするための接続オブジェクト
 * @author satoshi akabane@logical-paradox.org
 * @version $Revision: 1.2 $
 */
public class DomainConnection implements Comparable {
	/** ロガー */
	private static final Log log = LogFactory.getLog(DomainConnection.class);

	/** ドメインの接続状態(true:接続中 / false:切断中) */
	private boolean connection = false;
	/** この接続を共有しているクライアントノードIDのリスト */
	private List sharingClients = new ArrayList();
	/** 接続先ドメイン情報 */
	private Domain domain;
	/** 接続済みノードリスト(DacConnection) */
	private HashMap nodes = new HashMap();

	/**
	 * コンストラクタ
	 * @param nodeId この接続を共有するノードID
	 * @param dm 接続先ドメイン情報
	 */
	public DomainConnection(String nodeId, Domain dm) throws DomainAccessException {
		sharingClients.add(nodeId);
		domain = dm;

		// ドメイン接続
		connect();
	}
	/**
	 * ドメイン情報を更新する
	 * ドメイン情報を更新する際は、Dac接続などを気にする必要があることに注意する
	 * @param dm 新しいドメイン情報
	 */
	public void updateDomain(Domain dm) {
		domain = dm;
	}
	/**
	 * 関連付けられているドメイン情報を返す．
	 * @return ドメイン情報
	 */
	public Domain getDomain() {
		return domain;
	}
	/**
	 * 接続共有数を返す
	 * @return 接続共有数
	 */
	public int getSharedNum() {
		synchronized(sharingClients) {
			return sharingClients.size();
		}
	}
	/**
	 * 接続済みかどうかを返す
	 * @return true:接続済み / false:切断中
	 */
	public boolean isConnected() {
		return connection;
	}
	/**
	 * このドメイン接続にクライアントノードを割り当てる
	 * @param clientNodeId クライアントノードID
	 */
	public void share(String clientNodeId) {
		sharingClients.add(clientNodeId);
	}
	/**
	 * このドメイン接続からクライアントノードを除去する．
	 * @param clientNodeId クライアントノードID
	 */
	public void unshare(String clientNodeId) {
		sharingClients.remove(clientNodeId);
	}
	/**
	 * 指定されたクライアントに対して接続が割り当てられているかどうかを返す
	 * @param nodeId ノードID
	 * @return true:割り当て済み / false:未割り当て
	 */
	public boolean contains(String nodeId) {
		return sharingClients.contains(nodeId);
	}
	/**
	 * ドメインに参加している全てのノードに対して接続する
	 * 接続できないノードがあった場合、異常終了とする<br>
	 * @return ドメインに対して透過的にアクセスするためのセツゾクオブジェクト
	 * @throws DomainAccessException ドメインへの接続に失敗した
	 */
	protected void connect() throws DomainAccessException {
		log.info("ドメイン[" + domain.getDomainName() + "]に接続しています");

		boolean succeed = true;

		// ドメインに参加しているノードに接続する
		for(Iterator it = domain.getNodeList().keySet().iterator(); it.hasNext();) {
			String uri = (String)it.next();
			try {
				// ノードと接続する
				DacConnection con = DacConnection.open(uri);
				nodes.put(uri, con);
			} catch (IOException e) {
				log.error("ドメイン[" + domain.getDomainName() + "]のノード[" + uri + "]に接続できませんでした:", e);
				succeed = false;
			}
		}

		if(succeed == false) {
			log.error("ドメインのいくつかのノードに接続できなかったため、接続済みのノードと切断します");
			disconnect();
			connection = false;
			throw new DomainAccessException();
		} else {
			log.info("ドメイン[" + domain.getDomainName() + "]に接続しました");
			connection = true;
		}
	}
	/**
	 * ドメイン情報を破棄する
	 * ドメインに参加している全てのノードとの接続をクローズする
	 * エラーが発生しても無視して続行する
	 */
	public void disconnect() {
		log.info("ドメイン[" + domain.getDomainName() + "]を破棄しています");
		synchronized(nodes) {
			for(Iterator it = nodes.values().iterator(); it.hasNext();) {
				DacConnection node = (DacConnection)it.next();
				try {
					node.close();
				} catch(IOException e) {
					log.warn("ノード[" + node.getHost() + "]との接続切断処理中にエラーが発生しました:", e);
				}
			}

			// 接続済みノードをクリアする
			// エラーが発生して切断に失敗した接続についても，無視してクリアする
			// 結果として接続しっぱなしになったとしても，DACでタイムアウトするので問題ないかも
			nodes.clear();
		}
		connection = false;
		log.info("ドメイン[" + domain.getDomainName() + "]を破棄しました");
	}
	/**
	 * 指定クライアントノードを、このドメイン接続から離脱させる
	 * @param clientNodeId クライアントノートID
	 * @throws DomainAccessException 離脱に失敗
	 */
	public void logout(String clientNodeId) throws DomainAccessException {
		synchronized(sharingClients) {
			if(sharingClients.contains(clientNodeId) == false) {
				// そんなクライアントない
				throw new DomainAccessException();
			}
			sharingClients.remove(clientNodeId);
		}
	}
	/**
	 * ドメイン接続に対して指定ノードを追加する．
	 * 接続中のクライアントの状態などを意識して、安全にノードを追加する
	 * @param node ノード
	 */
	public void join(String node) {
		try {
			if(nodes.get(node) != null) {
				log.warn("ノード[" + node + "]との接続は既に確立していますので、無視します");
				return;
			}
			DacConnection con = DacConnection.open(node);
			synchronized(this) {
				nodes.put(node, con);
			}
			log.error("ノード[" + node + "]との接続を確立しました。");
		} catch (IOException e) {
			log.error("ノード[" + node + "]との接続確立に失敗しました。このノードは無視されます。", e);
		}

	}
	/**
	 * ドメイン接続から指定ノードを離脱させる．
	 * 接続中のクライアントの状態などを意識して、安全にノードを切り離す
	 * @param node ノード
	 */
	public void leave(String node) {
		synchronized(nodes) {
			DacConnection target = null;

			for(Iterator it = nodes.values().iterator(); it.hasNext();) {
				DacConnection con = (DacConnection)it.next();
				if(con.getHost().equals(node)) {
					// このオブジェクトを破棄
					target = con;
				}
			}

			if(target != null) {
				nodes.remove(target);
				try {
					target.close();
				} catch (IOException e) {
					log.error("ノード[" + node  + "]の離脱に失敗しました");
				}
			}
		}

	}
	/**
	 * ドメインに割り当てられているファイルサイズを返す
	 * @return ファイルサイズ
	 * @throws DomainAccessException ファイルサイズ取得に失敗した場合に発生する例外
	 */
	public long size() throws DomainAccessException {
		long size = 0;
		for(Iterator it = domain.getNodeList().values().iterator(); it.hasNext();) {
			DiskArrayNode node = (DiskArrayNode)it.next();
			size += node.getSize();
		}

		return size;
	}
	/**
	 * 所定位置からデータを書き込む
	 * @param recordBegin 書き込み開始位置
	 * @param bstream 書き込むデータ
	 * @throws DomainAccessException 書き込みに失敗した
	 */
	public void write(long recordBegin, byte[] bstream) throws DomainAccessException {
		DiskArrayNode node = select(recordBegin);
		if(node == null) {
			throw new DomainAccessException();
		}

		// 接続オブジェクトの選択
		DacConnection con = (DacConnection)nodes.get(node.getHost());
		if(con == null) {
			throw new DomainAccessException();
		}

		// 書き込み
		con.write(recordBegin, bstream);
	}
	/**
	 * 所定位置からデータを読み込む．
	 * 指定されたレコード番号を開始位置として、レコード単位で読み込む．
	 * @param recordBegin 読み込み開始位置
	 * @param bstream 読み込んだデータを格納するバッファ
	 * @return 読み込んだデータ数
	 */
	public int read(long recordBegin, byte[] bstream) throws DomainAccessException {
		DiskArrayNode node = select(recordBegin);
		if(node == null) {
			throw new DomainAccessException();
		}

		// 接続オブジェクトの選択
		DacConnection con = (DacConnection)nodes.get(node.getHost());
		if(con == null) {
			throw new DomainAccessException();
		}

		// 読み込み
		int records = bstream.length / domain.getRecordSize();
		if(bstream.length % domain.getRecordSize() > 0) {
			records++;
		}
		int readRecords = con.read(recordBegin, records, bstream);

		return readRecords;
	}
	/**
	 * 指定レコードを保持するノードを選択する。
	 * @param recordBegin 開始レコード番号
	 * @return ノード情報(null:なかった)
	 */
	protected DiskArrayNode select(long recordBegin) {
		for(Iterator it = domain.getNodeList().values().iterator(); it.hasNext();) {
			DiskArrayNode node = (DiskArrayNode)it.next();
			if(node.contains(recordBegin) == true) {
				return node;
			}
		}
		return null;
	}
	/**
	 * 接続共有数で比較する(ソート用)
	 * @param o 比較対象
	 * @return this ＜ o: -1 / this ＞ o : 1 / this == o : 0
	 */
	public int compareTo(Object o) {
		DomainConnection dest = (DomainConnection)o;

		if(dest.getSharedNum() == getSharedNum()) {
			return 0;
		}

		return getSharedNum() < dest.getSharedNum() ? -1 : 1;
	}
	/**
	 * 書き込むのに最適な位置を自動計算する．
	 * 書き込みたいデータが連続して書き込める位置を検索する．<br>
	 * 基本的には以下のアルゴリズムとなる：
	 * <ol>
	 * <li>各DACに対して、最適ストア領域を問い合わせる</li>
	 * <li>問い合わせ結果を集約し、最もレコード番号の小さいものを最適レコード番号とする</li>
	 * </ol>
	 * @param size 書き込みたいデータバイト数
	 * @return 最適シークポインタ(-1:もう空き領域が存在しない)
	 */
	public long computeBestSeekPointer(int size) {
		long lastRecord = -1L;

		// 問い合わせスレッド
		DacQueryThread[] threads = new DacQueryThread[domain.getNodeList().size()];

		int cnt = 0;
		for(Iterator it = domain.getNodeList().values().iterator(); it.hasNext(); cnt++) {
			DiskArrayNode node = (DiskArrayNode)it.next();
			DacConnection con = (DacConnection)nodes.get(node.getHost());
			DacQueryCommand command = new DacBestStoreSpaceQueryCommand(con, size);
			threads[cnt] = new DacQueryThread(command);

			threads[cnt].start();
		}

		// 並列処理での問い合わせ開始
		for(int i = 0; i < threads.length; i++) {
			try {
				threads[i].join();
			} catch(InterruptedException e) {
				// スレッド終了待ち合わせ中に発生した割り込み
				// 失敗したならそれはそれでもうどうでもいいので，そのまま無視する
			}
		}

		// 検索結果の取りまとめ
		for(int i = 0; i < threads.length; i++) {
			if(threads[i].getCausedException() == null) {
				Long found = (Long)threads[i].getResult();
				if(found != null && lastRecord > found.longValue()) {
					lastRecord = found.longValue();
				}
			} else {
				log.warn("最適ストア領域検索時に例外が発生しました。処理としては、該当DACがスキップされました：", threads[i].getCausedException());
			}
		}

		// 発見されたレコード番号をシークポインタに変換して返す
		if(lastRecord < 0) {
			return -1;
		} else {
			return lastRecord * domain.getRecordSize();
		}
	}
}
