You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cassandra.apache.org by jb...@apache.org on 2009/07/30 17:30:27 UTC
svn commit: r799331 [15/29] - in /incubator/cassandra/trunk: ./
src/java/org/apache/cassandra/concurrent/
src/java/org/apache/cassandra/config/ src/java/org/apache/cassandra/db/
src/java/org/apache/cassandra/dht/ src/java/org/apache/cassandra/gms/ src/...
Modified: incubator/cassandra/trunk/src/java/org/apache/cassandra/gms/HeartBeatState.java
URL: http://svn.apache.org/viewvc/incubator/cassandra/trunk/src/java/org/apache/cassandra/gms/HeartBeatState.java?rev=799331&r1=799330&r2=799331&view=diff
==============================================================================
--- incubator/cassandra/trunk/src/java/org/apache/cassandra/gms/HeartBeatState.java (original)
+++ incubator/cassandra/trunk/src/java/org/apache/cassandra/gms/HeartBeatState.java Thu Jul 30 15:30:21 2009
@@ -1,109 +1,109 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you 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.
- */
-
-package org.apache.cassandra.gms;
-
-import java.util.concurrent.atomic.AtomicInteger;
-import java.io.DataInputStream;
-import java.io.DataOutputStream;
-import java.io.IOException;
-
-import org.apache.cassandra.io.ICompactSerializer;
-
-
-/**
- * HeartBeat State associated with any given endpoint.
- * Author : Avinash Lakshman ( alakshman@facebook.com) & Prashant Malik ( pmalik@facebook.com )
- */
-
-class HeartBeatState
-{
- private static ICompactSerializer<HeartBeatState> serializer_;
-
- static
- {
- serializer_ = new HeartBeatStateSerializer();
- }
-
- int generation_;
- AtomicInteger heartbeat_;
- int version_;
-
- HeartBeatState()
- {
- }
-
- HeartBeatState(int generation, int heartbeat)
- {
- this(generation, heartbeat, 0);
- }
-
- HeartBeatState(int generation, int heartbeat, int version)
- {
- generation_ = generation;
- heartbeat_ = new AtomicInteger(heartbeat);
- version_ = version;
- }
-
- public static ICompactSerializer<HeartBeatState> serializer()
- {
- return serializer_;
- }
-
- int getGeneration()
- {
- return generation_;
- }
-
- void updateGeneration()
- {
- ++generation_;
- version_ = VersionGenerator.getNextVersion();
- }
-
- int getHeartBeat()
- {
- return heartbeat_.get();
- }
-
- void updateHeartBeat()
- {
- heartbeat_.incrementAndGet();
- version_ = VersionGenerator.getNextVersion();
- }
-
- int getHeartBeatVersion()
- {
- return version_;
- }
-};
-
-class HeartBeatStateSerializer implements ICompactSerializer<HeartBeatState>
-{
- public void serialize(HeartBeatState hbState, DataOutputStream dos) throws IOException
- {
- dos.writeInt(hbState.generation_);
- dos.writeInt(hbState.heartbeat_.get());
- dos.writeInt(hbState.version_);
- }
-
- public HeartBeatState deserialize(DataInputStream dis) throws IOException
- {
- return new HeartBeatState(dis.readInt(), dis.readInt(), dis.readInt());
- }
-}
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.
+ */
+
+package org.apache.cassandra.gms;
+
+import java.util.concurrent.atomic.AtomicInteger;
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+
+import org.apache.cassandra.io.ICompactSerializer;
+
+
+/**
+ * HeartBeat State associated with any given endpoint.
+ * Author : Avinash Lakshman ( alakshman@facebook.com) & Prashant Malik ( pmalik@facebook.com )
+ */
+
+class HeartBeatState
+{
+ private static ICompactSerializer<HeartBeatState> serializer_;
+
+ static
+ {
+ serializer_ = new HeartBeatStateSerializer();
+ }
+
+ int generation_;
+ AtomicInteger heartbeat_;
+ int version_;
+
+ HeartBeatState()
+ {
+ }
+
+ HeartBeatState(int generation, int heartbeat)
+ {
+ this(generation, heartbeat, 0);
+ }
+
+ HeartBeatState(int generation, int heartbeat, int version)
+ {
+ generation_ = generation;
+ heartbeat_ = new AtomicInteger(heartbeat);
+ version_ = version;
+ }
+
+ public static ICompactSerializer<HeartBeatState> serializer()
+ {
+ return serializer_;
+ }
+
+ int getGeneration()
+ {
+ return generation_;
+ }
+
+ void updateGeneration()
+ {
+ ++generation_;
+ version_ = VersionGenerator.getNextVersion();
+ }
+
+ int getHeartBeat()
+ {
+ return heartbeat_.get();
+ }
+
+ void updateHeartBeat()
+ {
+ heartbeat_.incrementAndGet();
+ version_ = VersionGenerator.getNextVersion();
+ }
+
+ int getHeartBeatVersion()
+ {
+ return version_;
+ }
+};
+
+class HeartBeatStateSerializer implements ICompactSerializer<HeartBeatState>
+{
+ public void serialize(HeartBeatState hbState, DataOutputStream dos) throws IOException
+ {
+ dos.writeInt(hbState.generation_);
+ dos.writeInt(hbState.heartbeat_.get());
+ dos.writeInt(hbState.version_);
+ }
+
+ public HeartBeatState deserialize(DataInputStream dis) throws IOException
+ {
+ return new HeartBeatState(dis.readInt(), dis.readInt(), dis.readInt());
+ }
+}
Modified: incubator/cassandra/trunk/src/java/org/apache/cassandra/gms/IEndPointStateChangePublisher.java
URL: http://svn.apache.org/viewvc/incubator/cassandra/trunk/src/java/org/apache/cassandra/gms/IEndPointStateChangePublisher.java?rev=799331&r1=799330&r2=799331&view=diff
==============================================================================
--- incubator/cassandra/trunk/src/java/org/apache/cassandra/gms/IEndPointStateChangePublisher.java (original)
+++ incubator/cassandra/trunk/src/java/org/apache/cassandra/gms/IEndPointStateChangePublisher.java Thu Jul 30 15:30:21 2009
@@ -1,41 +1,41 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you 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.
- */
-
-package org.apache.cassandra.gms;
-
-/**
- * This is implemented by the Gossiper module to publish change events to interested parties.
- * Interested parties register/unregister interest by invoking the methods of this interface.
- *
- * Author : Avinash Lakshman ( alakshman@facebook.com) & Prashant Malik ( pmalik@facebook.com )
- */
-
-public interface IEndPointStateChangePublisher
-{
- /**
- * Register for interesting state changes.
- * @param subcriber module which implements the IEndPointStateChangeSubscriber
- */
- public void register(IEndPointStateChangeSubscriber subcriber);
-
- /**
- * Unregister interest for state changes.
- * @param subcriber module which implements the IEndPointStateChangeSubscriber
- */
- public void unregister(IEndPointStateChangeSubscriber subcriber);
-}
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.
+ */
+
+package org.apache.cassandra.gms;
+
+/**
+ * This is implemented by the Gossiper module to publish change events to interested parties.
+ * Interested parties register/unregister interest by invoking the methods of this interface.
+ *
+ * Author : Avinash Lakshman ( alakshman@facebook.com) & Prashant Malik ( pmalik@facebook.com )
+ */
+
+public interface IEndPointStateChangePublisher
+{
+ /**
+ * Register for interesting state changes.
+ * @param subcriber module which implements the IEndPointStateChangeSubscriber
+ */
+ public void register(IEndPointStateChangeSubscriber subcriber);
+
+ /**
+ * Unregister interest for state changes.
+ * @param subcriber module which implements the IEndPointStateChangeSubscriber
+ */
+ public void unregister(IEndPointStateChangeSubscriber subcriber);
+}
Modified: incubator/cassandra/trunk/src/java/org/apache/cassandra/gms/IEndPointStateChangeSubscriber.java
URL: http://svn.apache.org/viewvc/incubator/cassandra/trunk/src/java/org/apache/cassandra/gms/IEndPointStateChangeSubscriber.java?rev=799331&r1=799330&r2=799331&view=diff
==============================================================================
--- incubator/cassandra/trunk/src/java/org/apache/cassandra/gms/IEndPointStateChangeSubscriber.java (original)
+++ incubator/cassandra/trunk/src/java/org/apache/cassandra/gms/IEndPointStateChangeSubscriber.java Thu Jul 30 15:30:21 2009
@@ -1,44 +1,44 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you 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.
- */
-
-package org.apache.cassandra.gms;
-
-import org.apache.cassandra.net.EndPoint;
-
-/**
- * This is called by an instance of the IEndPointStateChangePublisher to notify
- * interested parties about changes in the the state associated with any endpoint.
- * For instance if node A figures there is a changes in state for an endpoint B
- * it notifies all interested parties of this change. It is upto to the registered
- * instance to decide what he does with this change. Not all modules maybe interested
- * in all state changes.
- *
- * Author : Avinash Lakshman ( alakshman@facebook.com) & Prashant Malik ( pmalik@facebook.com )
- */
-
-public interface IEndPointStateChangeSubscriber
-{
- /**
- * Use to inform interested parties about the change in the state
- * for specified endpoint
- *
- * @param endpoint endpoint for which the state change occurred.
- * @param epState state that actually changed for the above endpoint.
- */
- public void onChange(EndPoint endpoint, EndPointState epState);
-}
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.
+ */
+
+package org.apache.cassandra.gms;
+
+import org.apache.cassandra.net.EndPoint;
+
+/**
+ * This is called by an instance of the IEndPointStateChangePublisher to notify
+ * interested parties about changes in the the state associated with any endpoint.
+ * For instance if node A figures there is a changes in state for an endpoint B
+ * it notifies all interested parties of this change. It is upto to the registered
+ * instance to decide what he does with this change. Not all modules maybe interested
+ * in all state changes.
+ *
+ * Author : Avinash Lakshman ( alakshman@facebook.com) & Prashant Malik ( pmalik@facebook.com )
+ */
+
+public interface IEndPointStateChangeSubscriber
+{
+ /**
+ * Use to inform interested parties about the change in the state
+ * for specified endpoint
+ *
+ * @param endpoint endpoint for which the state change occurred.
+ * @param epState state that actually changed for the above endpoint.
+ */
+ public void onChange(EndPoint endpoint, EndPointState epState);
+}
Modified: incubator/cassandra/trunk/src/java/org/apache/cassandra/gms/IFailureDetectionEventListener.java
URL: http://svn.apache.org/viewvc/incubator/cassandra/trunk/src/java/org/apache/cassandra/gms/IFailureDetectionEventListener.java?rev=799331&r1=799330&r2=799331&view=diff
==============================================================================
--- incubator/cassandra/trunk/src/java/org/apache/cassandra/gms/IFailureDetectionEventListener.java (original)
+++ incubator/cassandra/trunk/src/java/org/apache/cassandra/gms/IFailureDetectionEventListener.java Thu Jul 30 15:30:21 2009
@@ -1,44 +1,44 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you 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.
- */
-
-package org.apache.cassandra.gms;
-
-import org.apache.cassandra.net.EndPoint;
-
-/**
- * Implemented by the Gossiper to either convict/suspect an endpoint
- * based on the PHI calculated by the Failure Detector on the inter-arrival
- * times of the heart beats.
- *
- * Author : Avinash Lakshman ( alakshman@facebook.com) & Prashant Malik ( pmalik@facebook.com )
- */
-
-public interface IFailureDetectionEventListener
-{
- /**
- * Convict the specified endpoint.
- * @param ep endpoint to be convicted
- */
- public void convict(EndPoint ep);
-
- /**
- * Suspect the specified endpoint.
- * @param ep endpoint to be suspected.
- */
- public void suspect(EndPoint ep);
-}
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.
+ */
+
+package org.apache.cassandra.gms;
+
+import org.apache.cassandra.net.EndPoint;
+
+/**
+ * Implemented by the Gossiper to either convict/suspect an endpoint
+ * based on the PHI calculated by the Failure Detector on the inter-arrival
+ * times of the heart beats.
+ *
+ * Author : Avinash Lakshman ( alakshman@facebook.com) & Prashant Malik ( pmalik@facebook.com )
+ */
+
+public interface IFailureDetectionEventListener
+{
+ /**
+ * Convict the specified endpoint.
+ * @param ep endpoint to be convicted
+ */
+ public void convict(EndPoint ep);
+
+ /**
+ * Suspect the specified endpoint.
+ * @param ep endpoint to be suspected.
+ */
+ public void suspect(EndPoint ep);
+}
Modified: incubator/cassandra/trunk/src/java/org/apache/cassandra/gms/IFailureDetector.java
URL: http://svn.apache.org/viewvc/incubator/cassandra/trunk/src/java/org/apache/cassandra/gms/IFailureDetector.java?rev=799331&r1=799330&r2=799331&view=diff
==============================================================================
--- incubator/cassandra/trunk/src/java/org/apache/cassandra/gms/IFailureDetector.java (original)
+++ incubator/cassandra/trunk/src/java/org/apache/cassandra/gms/IFailureDetector.java Thu Jul 30 15:30:21 2009
@@ -1,72 +1,72 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you 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.
- */
-
-package org.apache.cassandra.gms;
-
-import org.apache.cassandra.net.EndPoint;
-
-/**
- * An interface that provides an application with the ability
- * to query liveness information of a node in the cluster. It
- * also exposes methods which help an application register callbacks
- * for notifications of liveness information of nodes.
- *
- * Author : Avinash Lakshman ( alakshman@facebook.com) & Prashant Malik ( pmalik@facebook.com )
- */
-
-public interface IFailureDetector
-{
- /**
- * Failure Detector's knowledge of whether a node is up or
- * down.
- *
- * @param ep endpoint in question.
- * @return true if UP and false if DOWN.
- */
- public boolean isAlive(EndPoint ep);
-
- /**
- * This method is invoked by any entity wanting to interrogate the status of an endpoint.
- * In our case it would be the Gossiper. The Failure Detector will then calculate Phi and
- * deem an endpoint as suspicious or alive as explained in the Hayashibara paper.
- *
- * param ep endpoint for which we interpret the inter arrival times.
- */
- public void interpret(EndPoint ep);
-
- /**
- * This method is invoked by the receiver of the heartbeat. In our case it would be
- * the Gossiper. Gossiper inform the Failure Detector on receipt of a heartbeat. The
- * FailureDetector will then sample the arrival time as explained in the paper.
- *
- * param ep endpoint being reported.
- */
- public void report(EndPoint ep);
-
- /**
- * Register interest for Failure Detector events.
- * @param listener implementation of an application provided IFailureDetectionEventListener
- */
- public void registerFailureDetectionEventListener(IFailureDetectionEventListener listener);
-
- /**
- * Un-register interest for Failure Detector events.
- * @param listener implementation of an application provided IFailureDetectionEventListener
- */
- public void unregisterFailureDetectionEventListener(IFailureDetectionEventListener listener);
-}
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.
+ */
+
+package org.apache.cassandra.gms;
+
+import org.apache.cassandra.net.EndPoint;
+
+/**
+ * An interface that provides an application with the ability
+ * to query liveness information of a node in the cluster. It
+ * also exposes methods which help an application register callbacks
+ * for notifications of liveness information of nodes.
+ *
+ * Author : Avinash Lakshman ( alakshman@facebook.com) & Prashant Malik ( pmalik@facebook.com )
+ */
+
+public interface IFailureDetector
+{
+ /**
+ * Failure Detector's knowledge of whether a node is up or
+ * down.
+ *
+ * @param ep endpoint in question.
+ * @return true if UP and false if DOWN.
+ */
+ public boolean isAlive(EndPoint ep);
+
+ /**
+ * This method is invoked by any entity wanting to interrogate the status of an endpoint.
+ * In our case it would be the Gossiper. The Failure Detector will then calculate Phi and
+ * deem an endpoint as suspicious or alive as explained in the Hayashibara paper.
+ *
+ * param ep endpoint for which we interpret the inter arrival times.
+ */
+ public void interpret(EndPoint ep);
+
+ /**
+ * This method is invoked by the receiver of the heartbeat. In our case it would be
+ * the Gossiper. Gossiper inform the Failure Detector on receipt of a heartbeat. The
+ * FailureDetector will then sample the arrival time as explained in the paper.
+ *
+ * param ep endpoint being reported.
+ */
+ public void report(EndPoint ep);
+
+ /**
+ * Register interest for Failure Detector events.
+ * @param listener implementation of an application provided IFailureDetectionEventListener
+ */
+ public void registerFailureDetectionEventListener(IFailureDetectionEventListener listener);
+
+ /**
+ * Un-register interest for Failure Detector events.
+ * @param listener implementation of an application provided IFailureDetectionEventListener
+ */
+ public void unregisterFailureDetectionEventListener(IFailureDetectionEventListener listener);
+}
Modified: incubator/cassandra/trunk/src/java/org/apache/cassandra/gms/IFailureNotification.java
URL: http://svn.apache.org/viewvc/incubator/cassandra/trunk/src/java/org/apache/cassandra/gms/IFailureNotification.java?rev=799331&r1=799330&r2=799331&view=diff
==============================================================================
--- incubator/cassandra/trunk/src/java/org/apache/cassandra/gms/IFailureNotification.java (original)
+++ incubator/cassandra/trunk/src/java/org/apache/cassandra/gms/IFailureNotification.java Thu Jul 30 15:30:21 2009
@@ -1,31 +1,31 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you 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.
- */
-
-package org.apache.cassandra.gms;
-
-import org.apache.cassandra.net.EndPoint;
-
-/**
- * Author : Avinash Lakshman ( alakshman@facebook.com) & Prashant Malik ( pmalik@facebook.com )
- */
-
-public interface IFailureNotification
-{
- public void suspect(EndPoint ep);
- public void revive(EndPoint ep);
-}
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.
+ */
+
+package org.apache.cassandra.gms;
+
+import org.apache.cassandra.net.EndPoint;
+
+/**
+ * Author : Avinash Lakshman ( alakshman@facebook.com) & Prashant Malik ( pmalik@facebook.com )
+ */
+
+public interface IFailureNotification
+{
+ public void suspect(EndPoint ep);
+ public void revive(EndPoint ep);
+}
Modified: incubator/cassandra/trunk/src/java/org/apache/cassandra/gms/JoinMessage.java
URL: http://svn.apache.org/viewvc/incubator/cassandra/trunk/src/java/org/apache/cassandra/gms/JoinMessage.java?rev=799331&r1=799330&r2=799331&view=diff
==============================================================================
--- incubator/cassandra/trunk/src/java/org/apache/cassandra/gms/JoinMessage.java (original)
+++ incubator/cassandra/trunk/src/java/org/apache/cassandra/gms/JoinMessage.java Thu Jul 30 15:30:21 2009
@@ -1,66 +1,66 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you 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.
- */
-
-package org.apache.cassandra.gms;
-
-import java.io.DataInputStream;
-import java.io.DataOutputStream;
-import java.io.IOException;
-import java.util.List;
-
-import org.apache.cassandra.io.ICompactSerializer;
-
-
-/**
- * Author : Avinash Lakshman ( alakshman@facebook.com) & Prashant Malik ( pmalik@facebook.com )
- */
-
-class JoinMessage
-{
- private static ICompactSerializer<JoinMessage> serializer_;
- static
- {
- serializer_ = new JoinMessageSerializer();
- }
-
- static ICompactSerializer<JoinMessage> serializer()
- {
- return serializer_;
- }
-
- String clusterId_;
-
- JoinMessage(String clusterId)
- {
- clusterId_ = clusterId;
- }
-}
-
-class JoinMessageSerializer implements ICompactSerializer<JoinMessage>
-{
- public void serialize(JoinMessage joinMessage, DataOutputStream dos) throws IOException
- {
- dos.writeUTF(joinMessage.clusterId_);
- }
-
- public JoinMessage deserialize(DataInputStream dis) throws IOException
- {
- String clusterId = dis.readUTF();
- return new JoinMessage(clusterId);
- }
-}
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.
+ */
+
+package org.apache.cassandra.gms;
+
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.util.List;
+
+import org.apache.cassandra.io.ICompactSerializer;
+
+
+/**
+ * Author : Avinash Lakshman ( alakshman@facebook.com) & Prashant Malik ( pmalik@facebook.com )
+ */
+
+class JoinMessage
+{
+ private static ICompactSerializer<JoinMessage> serializer_;
+ static
+ {
+ serializer_ = new JoinMessageSerializer();
+ }
+
+ static ICompactSerializer<JoinMessage> serializer()
+ {
+ return serializer_;
+ }
+
+ String clusterId_;
+
+ JoinMessage(String clusterId)
+ {
+ clusterId_ = clusterId;
+ }
+}
+
+class JoinMessageSerializer implements ICompactSerializer<JoinMessage>
+{
+ public void serialize(JoinMessage joinMessage, DataOutputStream dos) throws IOException
+ {
+ dos.writeUTF(joinMessage.clusterId_);
+ }
+
+ public JoinMessage deserialize(DataInputStream dis) throws IOException
+ {
+ String clusterId = dis.readUTF();
+ return new JoinMessage(clusterId);
+ }
+}
Modified: incubator/cassandra/trunk/src/java/org/apache/cassandra/gms/PureRandom.java
URL: http://svn.apache.org/viewvc/incubator/cassandra/trunk/src/java/org/apache/cassandra/gms/PureRandom.java?rev=799331&r1=799330&r2=799331&view=diff
==============================================================================
--- incubator/cassandra/trunk/src/java/org/apache/cassandra/gms/PureRandom.java (original)
+++ incubator/cassandra/trunk/src/java/org/apache/cassandra/gms/PureRandom.java Thu Jul 30 15:30:21 2009
@@ -1,83 +1,83 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you 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.
- */
-
-package org.apache.cassandra.gms;
-
-import java.util.Random;
-import java.util.BitSet;
-
-
-/**
- * Implementation of a PureRandomNumber generator. Use this class cautiously. Not
- * for general purpose use. Currently this is used by the Gossiper to choose a random
- * endpoint to Gossip to.
- *
- * Author : Avinash Lakshman ( alakshman@facebook.com) & Prashant Malik ( pmalik@facebook.com )
- */
-
-class PureRandom extends Random
-{
- private BitSet bs_ = new BitSet();
- private int lastUb_;
-
- PureRandom()
- {
- super();
- }
-
- public int nextInt(int ub)
- {
- if (ub <= 0)
- throw new IllegalArgumentException("ub must be positive");
-
- if ( lastUb_ != ub )
- {
- bs_.clear();
- lastUb_ = ub;
- }
- else if(bs_.cardinality() == ub)
- {
- bs_.clear();
- }
-
- int value = super.nextInt(ub);
- while ( bs_.get(value) )
- {
- value = super.nextInt(ub);
- }
- bs_.set(value);
- return value;
- }
-
- public static void main(String[] args) throws Throwable
- {
- Random pr = new PureRandom();
- int ubs[] = new int[] { 2, 3, 1, 10, 5, 0};
-
- for (int ub : ubs)
- {
- System.out.println("UB: " + String.valueOf(ub));
- for (int j = 0; j < 10; j++)
- {
- int junk = pr.nextInt(ub);
- // Do something with junk so JVM doesn't optimize away
- System.out.println(junk);
- }
- }
- }
-}
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.
+ */
+
+package org.apache.cassandra.gms;
+
+import java.util.Random;
+import java.util.BitSet;
+
+
+/**
+ * Implementation of a PureRandomNumber generator. Use this class cautiously. Not
+ * for general purpose use. Currently this is used by the Gossiper to choose a random
+ * endpoint to Gossip to.
+ *
+ * Author : Avinash Lakshman ( alakshman@facebook.com) & Prashant Malik ( pmalik@facebook.com )
+ */
+
+class PureRandom extends Random
+{
+ private BitSet bs_ = new BitSet();
+ private int lastUb_;
+
+ PureRandom()
+ {
+ super();
+ }
+
+ public int nextInt(int ub)
+ {
+ if (ub <= 0)
+ throw new IllegalArgumentException("ub must be positive");
+
+ if ( lastUb_ != ub )
+ {
+ bs_.clear();
+ lastUb_ = ub;
+ }
+ else if(bs_.cardinality() == ub)
+ {
+ bs_.clear();
+ }
+
+ int value = super.nextInt(ub);
+ while ( bs_.get(value) )
+ {
+ value = super.nextInt(ub);
+ }
+ bs_.set(value);
+ return value;
+ }
+
+ public static void main(String[] args) throws Throwable
+ {
+ Random pr = new PureRandom();
+ int ubs[] = new int[] { 2, 3, 1, 10, 5, 0};
+
+ for (int ub : ubs)
+ {
+ System.out.println("UB: " + String.valueOf(ub));
+ for (int j = 0; j < 10; j++)
+ {
+ int junk = pr.nextInt(ub);
+ // Do something with junk so JVM doesn't optimize away
+ System.out.println(junk);
+ }
+ }
+ }
+}
Modified: incubator/cassandra/trunk/src/java/org/apache/cassandra/gms/VersionGenerator.java
URL: http://svn.apache.org/viewvc/incubator/cassandra/trunk/src/java/org/apache/cassandra/gms/VersionGenerator.java?rev=799331&r1=799330&r2=799331&view=diff
==============================================================================
--- incubator/cassandra/trunk/src/java/org/apache/cassandra/gms/VersionGenerator.java (original)
+++ incubator/cassandra/trunk/src/java/org/apache/cassandra/gms/VersionGenerator.java Thu Jul 30 15:30:21 2009
@@ -1,37 +1,37 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you 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.
- */
-
-package org.apache.cassandra.gms;
-
-import java.util.concurrent.atomic.AtomicInteger;
-
-/**
- * A unique version number generator for any state that is generated by the
- * local node.
- * Author : Avinash Lakshman ( alakshman@facebook.com) & Prashant Malik ( pmalik@facebook.com )
- */
-
-public class VersionGenerator
-{
- private static AtomicInteger version_ = new AtomicInteger(0);
-
- public static int getNextVersion()
- {
- return version_.incrementAndGet();
- }
-}
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.
+ */
+
+package org.apache.cassandra.gms;
+
+import java.util.concurrent.atomic.AtomicInteger;
+
+/**
+ * A unique version number generator for any state that is generated by the
+ * local node.
+ * Author : Avinash Lakshman ( alakshman@facebook.com) & Prashant Malik ( pmalik@facebook.com )
+ */
+
+public class VersionGenerator
+{
+ private static AtomicInteger version_ = new AtomicInteger(0);
+
+ public static int getNextVersion()
+ {
+ return version_.incrementAndGet();
+ }
+}
Modified: incubator/cassandra/trunk/src/java/org/apache/cassandra/io/BufferedRandomAccessFile.java
URL: http://svn.apache.org/viewvc/incubator/cassandra/trunk/src/java/org/apache/cassandra/io/BufferedRandomAccessFile.java?rev=799331&r1=799330&r2=799331&view=diff
==============================================================================
--- incubator/cassandra/trunk/src/java/org/apache/cassandra/io/BufferedRandomAccessFile.java (original)
+++ incubator/cassandra/trunk/src/java/org/apache/cassandra/io/BufferedRandomAccessFile.java Thu Jul 30 15:30:21 2009
@@ -1,375 +1,375 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you 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.
- */
-
-package org.apache.cassandra.io;
-
-import java.io.File;
-import java.io.FileNotFoundException;
-import java.io.IOException;
-import java.io.RandomAccessFile;
-import java.util.Arrays;
-
-/**
- * A <code>BufferedRandomAccessFile</code> is like a
- * <code>RandomAccessFile</code>, but it uses a private buffer so that most
- * operations do not require a disk access.
- * <P>
- *
- * Note: The operations on this class are unmonitored. Also, the correct
- * functioning of the <code>RandomAccessFile</code> methods that are not
- * overridden here relies on the implementation of those methods in the
- * superclass.
- * Author : Avinash Lakshman ( alakshman@facebook.com) & Prashant Malik ( pmalik@facebook.com )
- */
-
-public final class BufferedRandomAccessFile extends RandomAccessFile
-{
- static final int LogBuffSz_ = 16; // 64K buffer
- public static final int BuffSz_ = (1 << LogBuffSz_);
- static final long BuffMask_ = ~(((long) BuffSz_) - 1L);
-
- /*
- * This implementation is based on the buffer implementation in Modula-3's
- * "Rd", "Wr", "RdClass", and "WrClass" interfaces.
- */
- private boolean dirty_; // true iff unflushed bytes exist
- private long curr_; // current position in file
- private long lo_, hi_; // bounds on characters in "buff"
- private byte[] buff_; // local buffer
- private long maxHi_; // this.lo + this.buff.length
- private boolean hitEOF_; // buffer contains last file block?
- private long diskPos_; // disk position
-
- /*
- * To describe the above fields, we introduce the following abstractions for
- * the file "f":
- *
- * len(f) the length of the file curr(f) the current position in the file
- * c(f) the abstract contents of the file disk(f) the contents of f's
- * backing disk file closed(f) true iff the file is closed
- *
- * "curr(f)" is an index in the closed interval [0, len(f)]. "c(f)" is a
- * character sequence of length "len(f)". "c(f)" and "disk(f)" may differ if
- * "c(f)" contains unflushed writes not reflected in "disk(f)". The flush
- * operation has the effect of making "disk(f)" identical to "c(f)".
- *
- * A file is said to be *valid* if the following conditions hold:
- *
- * V1. The "closed" and "curr" fields are correct:
- *
- * f.closed == closed(f) f.curr == curr(f)
- *
- * V2. The current position is either contained in the buffer, or just past
- * the buffer:
- *
- * f.lo <= f.curr <= f.hi
- *
- * V3. Any (possibly) unflushed characters are stored in "f.buff":
- *
- * (forall i in [f.lo, f.curr): c(f)[i] == f.buff[i - f.lo])
- *
- * V4. For all characters not covered by V3, c(f) and disk(f) agree:
- *
- * (forall i in [f.lo, len(f)): i not in [f.lo, f.curr) => c(f)[i] ==
- * disk(f)[i])
- *
- * V5. "f.dirty" is true iff the buffer contains bytes that should be
- * flushed to the file; by V3 and V4, only part of the buffer can be dirty.
- *
- * f.dirty == (exists i in [f.lo, f.curr): c(f)[i] != f.buff[i - f.lo])
- *
- * V6. this.maxHi == this.lo + this.buff.length
- *
- * Note that "f.buff" can be "null" in a valid file, since the range of
- * characters in V3 is empty when "f.lo == f.curr".
- *
- * A file is said to be *ready* if the buffer contains the current position,
- * i.e., when:
- *
- * R1. !f.closed && f.buff != null && f.lo <= f.curr && f.curr < f.hi
- *
- * When a file is ready, reading or writing a single byte can be performed
- * by reading or writing the in-memory buffer without performing a disk
- * operation.
- */
-
- /**
- * Open a new <code>BufferedRandomAccessFile</code> on <code>file</code>
- * in mode <code>mode</code>, which should be "r" for reading only, or
- * "rw" for reading and writing.
- */
- public BufferedRandomAccessFile(File file, String mode) throws IOException
- {
- super(file, mode);
- this.init(0);
- }
-
- public BufferedRandomAccessFile(File file, String mode, int size) throws IOException
- {
- super(file, mode);
- this.init(size);
- }
-
- /**
- * Open a new <code>BufferedRandomAccessFile</code> on the file named
- * <code>name</code> in mode <code>mode</code>, which should be "r" for
- * reading only, or "rw" for reading and writing.
- */
- public BufferedRandomAccessFile(String name, String mode) throws IOException
- {
- super(name, mode);
- this.init(0);
- }
-
- public BufferedRandomAccessFile(String name, String mode, int size) throws FileNotFoundException
- {
- super(name, mode);
- this.init(size);
- }
-
- private void init(int size)
- {
- this.dirty_ = false;
- this.lo_ = this.curr_ = this.hi_ = 0;
- this.buff_ = (size > BuffSz_) ? new byte[size] : new byte[BuffSz_];
- this.maxHi_ = (long) BuffSz_;
- this.hitEOF_ = false;
- this.diskPos_ = 0L;
- }
-
- public void close() throws IOException
- {
- this.flush();
- this.buff_ = null;
- super.close();
- }
-
- /**
- * Flush any bytes in the file's buffer that have not yet been written to
- * disk. If the file was created read-only, this method is a no-op.
- */
- public void flush() throws IOException
- {
- this.flushBuffer();
- }
-
- /* Flush any dirty bytes in the buffer to disk. */
- private void flushBuffer() throws IOException
- {
- if (this.dirty_)
- {
- if (this.diskPos_ != this.lo_)
- super.seek(this.lo_);
- int len = (int) (this.curr_ - this.lo_);
- super.write(this.buff_, 0, len);
- this.diskPos_ = this.curr_;
- this.dirty_ = false;
- }
- }
-
- /*
- * Read at most "this.buff.length" bytes into "this.buff", returning the
- * number of bytes read. If the return result is less than
- * "this.buff.length", then EOF was read.
- */
- private int fillBuffer() throws IOException
- {
- int cnt = 0;
- int rem = this.buff_.length;
- while (rem > 0)
- {
- int n = super.read(this.buff_, cnt, rem);
- if (n < 0)
- break;
- cnt += n;
- rem -= n;
- }
- if ( (cnt < 0) && (this.hitEOF_ = (cnt < this.buff_.length)) )
- {
- // make sure buffer that wasn't read is initialized with -1
- Arrays.fill(this.buff_, cnt, this.buff_.length, (byte) 0xff);
- }
- this.diskPos_ += cnt;
- return cnt;
- }
-
- /*
- * This method positions <code>this.curr</code> at position <code>pos</code>.
- * If <code>pos</code> does not fall in the current buffer, it flushes the
- * current buffer and loads the correct one.<p>
- *
- * On exit from this routine <code>this.curr == this.hi</code> iff <code>pos</code>
- * is at or past the end-of-file, which can only happen if the file was
- * opened in read-only mode.
- */
- public void seek(long pos) throws IOException
- {
- if (pos >= this.hi_ || pos < this.lo_)
- {
- // seeking outside of current buffer -- flush and read
- this.flushBuffer();
- this.lo_ = pos & BuffMask_; // start at BuffSz boundary
- this.maxHi_ = this.lo_ + (long) this.buff_.length;
- if (this.diskPos_ != this.lo_)
- {
- super.seek(this.lo_);
- this.diskPos_ = this.lo_;
- }
- int n = this.fillBuffer();
- this.hi_ = this.lo_ + (long) n;
- }
- else
- {
- // seeking inside current buffer -- no read required
- if (pos < this.curr_)
- {
- // if seeking backwards, we must flush to maintain V4
- this.flushBuffer();
- }
- }
- this.curr_ = pos;
- }
-
- public long getFilePointer()
- {
- return this.curr_;
- }
-
- public long length() throws IOException
- {
- // max accounts for the case where we have written past the old file length, but not yet flushed our buffer
- return Math.max(this.curr_, super.length());
- }
-
- public int read() throws IOException
- {
- if (this.curr_ >= this.hi_)
- {
- // test for EOF
- // if (this.hi < this.maxHi) return -1;
- if (this.hitEOF_)
- return -1;
-
- // slow path -- read another buffer
- this.seek(this.curr_);
- if (this.curr_ == this.hi_)
- return -1;
- }
- byte res = this.buff_[(int) (this.curr_ - this.lo_)];
- this.curr_++;
- return ((int) res) & 0xFF; // convert byte -> int
- }
-
- public int read(byte[] b) throws IOException
- {
- return this.read(b, 0, b.length);
- }
-
- public int read(byte[] b, int off, int len) throws IOException
- {
- if (this.curr_ >= this.hi_)
- {
- // test for EOF
- // if (this.hi < this.maxHi) return -1;
- if (this.hitEOF_)
- return -1;
-
- // slow path -- read another buffer
- this.seek(this.curr_);
- if (this.curr_ == this.hi_)
- return -1;
- }
- len = Math.min(len, (int) (this.hi_ - this.curr_));
- int buffOff = (int) (this.curr_ - this.lo_);
- System.arraycopy(this.buff_, buffOff, b, off, len);
- this.curr_ += len;
- return len;
- }
-
- public void write(int b) throws IOException
- {
- if (this.curr_ >= this.hi_)
- {
- if (this.hitEOF_ && this.hi_ < this.maxHi_)
- {
- // at EOF -- bump "hi"
- this.hi_++;
- }
- else
- {
- // slow path -- write current buffer; read next one
- this.seek(this.curr_);
- if (this.curr_ == this.hi_)
- {
- // appending to EOF -- bump "hi"
- this.hi_++;
- }
- }
- }
- this.buff_[(int) (this.curr_ - this.lo_)] = (byte) b;
- this.curr_++;
- this.dirty_ = true;
- }
-
- public void write(byte[] b) throws IOException
- {
- this.write(b, 0, b.length);
- }
-
- public void write(byte[] b, int off, int len) throws IOException
- {
- while (len > 0)
- {
- int n = this.writeAtMost(b, off, len);
- off += n;
- len -= n;
- this.dirty_ = true;
- }
- }
-
- /*
- * Write at most "len" bytes to "b" starting at position "off", and return
- * the number of bytes written.
- */
- private int writeAtMost(byte[] b, int off, int len) throws IOException
- {
- if (this.curr_ >= this.hi_)
- {
- if (this.hitEOF_ && this.hi_ < this.maxHi_)
- {
- // at EOF -- bump "hi"
- this.hi_ = this.maxHi_;
- }
- else
- {
- // slow path -- write current buffer; read next one
- this.seek(this.curr_);
- if (this.curr_ == this.hi_)
- {
- // appending to EOF -- bump "hi"
- this.hi_ = this.maxHi_;
- }
- }
- }
- len = Math.min(len, (int) (this.hi_ - this.curr_));
- int buffOff = (int) (this.curr_ - this.lo_);
- System.arraycopy(b, off, this.buff_, buffOff, len);
- this.curr_ += len;
- return len;
- }
-}
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.
+ */
+
+package org.apache.cassandra.io;
+
+import java.io.File;
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.io.RandomAccessFile;
+import java.util.Arrays;
+
+/**
+ * A <code>BufferedRandomAccessFile</code> is like a
+ * <code>RandomAccessFile</code>, but it uses a private buffer so that most
+ * operations do not require a disk access.
+ * <P>
+ *
+ * Note: The operations on this class are unmonitored. Also, the correct
+ * functioning of the <code>RandomAccessFile</code> methods that are not
+ * overridden here relies on the implementation of those methods in the
+ * superclass.
+ * Author : Avinash Lakshman ( alakshman@facebook.com) & Prashant Malik ( pmalik@facebook.com )
+ */
+
+public final class BufferedRandomAccessFile extends RandomAccessFile
+{
+ static final int LogBuffSz_ = 16; // 64K buffer
+ public static final int BuffSz_ = (1 << LogBuffSz_);
+ static final long BuffMask_ = ~(((long) BuffSz_) - 1L);
+
+ /*
+ * This implementation is based on the buffer implementation in Modula-3's
+ * "Rd", "Wr", "RdClass", and "WrClass" interfaces.
+ */
+ private boolean dirty_; // true iff unflushed bytes exist
+ private long curr_; // current position in file
+ private long lo_, hi_; // bounds on characters in "buff"
+ private byte[] buff_; // local buffer
+ private long maxHi_; // this.lo + this.buff.length
+ private boolean hitEOF_; // buffer contains last file block?
+ private long diskPos_; // disk position
+
+ /*
+ * To describe the above fields, we introduce the following abstractions for
+ * the file "f":
+ *
+ * len(f) the length of the file curr(f) the current position in the file
+ * c(f) the abstract contents of the file disk(f) the contents of f's
+ * backing disk file closed(f) true iff the file is closed
+ *
+ * "curr(f)" is an index in the closed interval [0, len(f)]. "c(f)" is a
+ * character sequence of length "len(f)". "c(f)" and "disk(f)" may differ if
+ * "c(f)" contains unflushed writes not reflected in "disk(f)". The flush
+ * operation has the effect of making "disk(f)" identical to "c(f)".
+ *
+ * A file is said to be *valid* if the following conditions hold:
+ *
+ * V1. The "closed" and "curr" fields are correct:
+ *
+ * f.closed == closed(f) f.curr == curr(f)
+ *
+ * V2. The current position is either contained in the buffer, or just past
+ * the buffer:
+ *
+ * f.lo <= f.curr <= f.hi
+ *
+ * V3. Any (possibly) unflushed characters are stored in "f.buff":
+ *
+ * (forall i in [f.lo, f.curr): c(f)[i] == f.buff[i - f.lo])
+ *
+ * V4. For all characters not covered by V3, c(f) and disk(f) agree:
+ *
+ * (forall i in [f.lo, len(f)): i not in [f.lo, f.curr) => c(f)[i] ==
+ * disk(f)[i])
+ *
+ * V5. "f.dirty" is true iff the buffer contains bytes that should be
+ * flushed to the file; by V3 and V4, only part of the buffer can be dirty.
+ *
+ * f.dirty == (exists i in [f.lo, f.curr): c(f)[i] != f.buff[i - f.lo])
+ *
+ * V6. this.maxHi == this.lo + this.buff.length
+ *
+ * Note that "f.buff" can be "null" in a valid file, since the range of
+ * characters in V3 is empty when "f.lo == f.curr".
+ *
+ * A file is said to be *ready* if the buffer contains the current position,
+ * i.e., when:
+ *
+ * R1. !f.closed && f.buff != null && f.lo <= f.curr && f.curr < f.hi
+ *
+ * When a file is ready, reading or writing a single byte can be performed
+ * by reading or writing the in-memory buffer without performing a disk
+ * operation.
+ */
+
+ /**
+ * Open a new <code>BufferedRandomAccessFile</code> on <code>file</code>
+ * in mode <code>mode</code>, which should be "r" for reading only, or
+ * "rw" for reading and writing.
+ */
+ public BufferedRandomAccessFile(File file, String mode) throws IOException
+ {
+ super(file, mode);
+ this.init(0);
+ }
+
+ public BufferedRandomAccessFile(File file, String mode, int size) throws IOException
+ {
+ super(file, mode);
+ this.init(size);
+ }
+
+ /**
+ * Open a new <code>BufferedRandomAccessFile</code> on the file named
+ * <code>name</code> in mode <code>mode</code>, which should be "r" for
+ * reading only, or "rw" for reading and writing.
+ */
+ public BufferedRandomAccessFile(String name, String mode) throws IOException
+ {
+ super(name, mode);
+ this.init(0);
+ }
+
+ public BufferedRandomAccessFile(String name, String mode, int size) throws FileNotFoundException
+ {
+ super(name, mode);
+ this.init(size);
+ }
+
+ private void init(int size)
+ {
+ this.dirty_ = false;
+ this.lo_ = this.curr_ = this.hi_ = 0;
+ this.buff_ = (size > BuffSz_) ? new byte[size] : new byte[BuffSz_];
+ this.maxHi_ = (long) BuffSz_;
+ this.hitEOF_ = false;
+ this.diskPos_ = 0L;
+ }
+
+ public void close() throws IOException
+ {
+ this.flush();
+ this.buff_ = null;
+ super.close();
+ }
+
+ /**
+ * Flush any bytes in the file's buffer that have not yet been written to
+ * disk. If the file was created read-only, this method is a no-op.
+ */
+ public void flush() throws IOException
+ {
+ this.flushBuffer();
+ }
+
+ /* Flush any dirty bytes in the buffer to disk. */
+ private void flushBuffer() throws IOException
+ {
+ if (this.dirty_)
+ {
+ if (this.diskPos_ != this.lo_)
+ super.seek(this.lo_);
+ int len = (int) (this.curr_ - this.lo_);
+ super.write(this.buff_, 0, len);
+ this.diskPos_ = this.curr_;
+ this.dirty_ = false;
+ }
+ }
+
+ /*
+ * Read at most "this.buff.length" bytes into "this.buff", returning the
+ * number of bytes read. If the return result is less than
+ * "this.buff.length", then EOF was read.
+ */
+ private int fillBuffer() throws IOException
+ {
+ int cnt = 0;
+ int rem = this.buff_.length;
+ while (rem > 0)
+ {
+ int n = super.read(this.buff_, cnt, rem);
+ if (n < 0)
+ break;
+ cnt += n;
+ rem -= n;
+ }
+ if ( (cnt < 0) && (this.hitEOF_ = (cnt < this.buff_.length)) )
+ {
+ // make sure buffer that wasn't read is initialized with -1
+ Arrays.fill(this.buff_, cnt, this.buff_.length, (byte) 0xff);
+ }
+ this.diskPos_ += cnt;
+ return cnt;
+ }
+
+ /*
+ * This method positions <code>this.curr</code> at position <code>pos</code>.
+ * If <code>pos</code> does not fall in the current buffer, it flushes the
+ * current buffer and loads the correct one.<p>
+ *
+ * On exit from this routine <code>this.curr == this.hi</code> iff <code>pos</code>
+ * is at or past the end-of-file, which can only happen if the file was
+ * opened in read-only mode.
+ */
+ public void seek(long pos) throws IOException
+ {
+ if (pos >= this.hi_ || pos < this.lo_)
+ {
+ // seeking outside of current buffer -- flush and read
+ this.flushBuffer();
+ this.lo_ = pos & BuffMask_; // start at BuffSz boundary
+ this.maxHi_ = this.lo_ + (long) this.buff_.length;
+ if (this.diskPos_ != this.lo_)
+ {
+ super.seek(this.lo_);
+ this.diskPos_ = this.lo_;
+ }
+ int n = this.fillBuffer();
+ this.hi_ = this.lo_ + (long) n;
+ }
+ else
+ {
+ // seeking inside current buffer -- no read required
+ if (pos < this.curr_)
+ {
+ // if seeking backwards, we must flush to maintain V4
+ this.flushBuffer();
+ }
+ }
+ this.curr_ = pos;
+ }
+
+ public long getFilePointer()
+ {
+ return this.curr_;
+ }
+
+ public long length() throws IOException
+ {
+ // max accounts for the case where we have written past the old file length, but not yet flushed our buffer
+ return Math.max(this.curr_, super.length());
+ }
+
+ public int read() throws IOException
+ {
+ if (this.curr_ >= this.hi_)
+ {
+ // test for EOF
+ // if (this.hi < this.maxHi) return -1;
+ if (this.hitEOF_)
+ return -1;
+
+ // slow path -- read another buffer
+ this.seek(this.curr_);
+ if (this.curr_ == this.hi_)
+ return -1;
+ }
+ byte res = this.buff_[(int) (this.curr_ - this.lo_)];
+ this.curr_++;
+ return ((int) res) & 0xFF; // convert byte -> int
+ }
+
+ public int read(byte[] b) throws IOException
+ {
+ return this.read(b, 0, b.length);
+ }
+
+ public int read(byte[] b, int off, int len) throws IOException
+ {
+ if (this.curr_ >= this.hi_)
+ {
+ // test for EOF
+ // if (this.hi < this.maxHi) return -1;
+ if (this.hitEOF_)
+ return -1;
+
+ // slow path -- read another buffer
+ this.seek(this.curr_);
+ if (this.curr_ == this.hi_)
+ return -1;
+ }
+ len = Math.min(len, (int) (this.hi_ - this.curr_));
+ int buffOff = (int) (this.curr_ - this.lo_);
+ System.arraycopy(this.buff_, buffOff, b, off, len);
+ this.curr_ += len;
+ return len;
+ }
+
+ public void write(int b) throws IOException
+ {
+ if (this.curr_ >= this.hi_)
+ {
+ if (this.hitEOF_ && this.hi_ < this.maxHi_)
+ {
+ // at EOF -- bump "hi"
+ this.hi_++;
+ }
+ else
+ {
+ // slow path -- write current buffer; read next one
+ this.seek(this.curr_);
+ if (this.curr_ == this.hi_)
+ {
+ // appending to EOF -- bump "hi"
+ this.hi_++;
+ }
+ }
+ }
+ this.buff_[(int) (this.curr_ - this.lo_)] = (byte) b;
+ this.curr_++;
+ this.dirty_ = true;
+ }
+
+ public void write(byte[] b) throws IOException
+ {
+ this.write(b, 0, b.length);
+ }
+
+ public void write(byte[] b, int off, int len) throws IOException
+ {
+ while (len > 0)
+ {
+ int n = this.writeAtMost(b, off, len);
+ off += n;
+ len -= n;
+ this.dirty_ = true;
+ }
+ }
+
+ /*
+ * Write at most "len" bytes to "b" starting at position "off", and return
+ * the number of bytes written.
+ */
+ private int writeAtMost(byte[] b, int off, int len) throws IOException
+ {
+ if (this.curr_ >= this.hi_)
+ {
+ if (this.hitEOF_ && this.hi_ < this.maxHi_)
+ {
+ // at EOF -- bump "hi"
+ this.hi_ = this.maxHi_;
+ }
+ else
+ {
+ // slow path -- write current buffer; read next one
+ this.seek(this.curr_);
+ if (this.curr_ == this.hi_)
+ {
+ // appending to EOF -- bump "hi"
+ this.hi_ = this.maxHi_;
+ }
+ }
+ }
+ len = Math.min(len, (int) (this.hi_ - this.curr_));
+ int buffOff = (int) (this.curr_ - this.lo_);
+ System.arraycopy(b, off, this.buff_, buffOff, len);
+ this.curr_ += len;
+ return len;
+ }
+}
Modified: incubator/cassandra/trunk/src/java/org/apache/cassandra/io/Coordinate.java
URL: http://svn.apache.org/viewvc/incubator/cassandra/trunk/src/java/org/apache/cassandra/io/Coordinate.java?rev=799331&r1=799330&r2=799331&view=diff
==============================================================================
--- incubator/cassandra/trunk/src/java/org/apache/cassandra/io/Coordinate.java (original)
+++ incubator/cassandra/trunk/src/java/org/apache/cassandra/io/Coordinate.java Thu Jul 30 15:30:21 2009
@@ -1,42 +1,42 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you 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.
- */
-package org.apache.cassandra.io;
-
-/**
- * Section of a file that needs to be scanned
- * is represented by this class.
-*/
-public class Coordinate
-{
- public final long start_;
- public final long end_;
-
- Coordinate(long start, long end)
- {
- start_ = start;
- end_ = end;
- }
-
- public String toString()
- {
- return "Coordinate(" +
- "start_=" + start_ +
- ", end_=" + end_ +
- ')';
- }
-}
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.
+ */
+package org.apache.cassandra.io;
+
+/**
+ * Section of a file that needs to be scanned
+ * is represented by this class.
+*/
+public class Coordinate
+{
+ public final long start_;
+ public final long end_;
+
+ Coordinate(long start, long end)
+ {
+ start_ = start;
+ end_ = end;
+ }
+
+ public String toString()
+ {
+ return "Coordinate(" +
+ "start_=" + start_ +
+ ", end_=" + end_ +
+ ')';
+ }
+}
Modified: incubator/cassandra/trunk/src/java/org/apache/cassandra/io/DataInputBuffer.java
URL: http://svn.apache.org/viewvc/incubator/cassandra/trunk/src/java/org/apache/cassandra/io/DataInputBuffer.java?rev=799331&r1=799330&r2=799331&view=diff
==============================================================================
--- incubator/cassandra/trunk/src/java/org/apache/cassandra/io/DataInputBuffer.java (original)
+++ incubator/cassandra/trunk/src/java/org/apache/cassandra/io/DataInputBuffer.java Thu Jul 30 15:30:21 2009
@@ -1,100 +1,100 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you 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.
- */
-
-package org.apache.cassandra.io;
-
-import java.io.*;
-
-
-/**
- * An implementation of the DataInputStream interface. This instance is completely thread
- * unsafe.
- *
- * Author : Avinash Lakshman ( alakshman@facebook.com) & Prashant Malik ( pmalik@facebook.com )
- */
-
-public final class DataInputBuffer extends DataInputStream
-{
- private static class Buffer extends ByteArrayInputStream
- {
- public Buffer()
- {
- super(new byte[] {});
- }
-
- public void reset(byte[] input, int start, int length)
- {
- this.buf = input;
- this.count = start + length;
- this.mark = start;
- this.pos = start;
- }
-
- public int getPosition()
- {
- return pos;
- }
-
- public void setPosition(int position)
- {
- pos = position;
- }
-
- public int getLength()
- {
- return count;
- }
- }
-
- private Buffer buffer_;
-
- /** Constructs a new empty buffer. */
- public DataInputBuffer()
- {
- this(new Buffer());
- }
-
- private DataInputBuffer(Buffer buffer)
- {
- super(buffer);
- this.buffer_ = buffer;
- }
-
- /** Resets the data that the buffer reads. */
- public void reset(byte[] input, int length)
- {
- buffer_.reset(input, 0, length);
- }
-
- /** Resets the data that the buffer reads. */
- public void reset(byte[] input, int start, int length)
- {
- buffer_.reset(input, start, length);
- }
-
- /** Returns the length of the input. */
- public int getLength()
- {
- return buffer_.getLength();
- }
-
- public int getPosition()
- {
- return buffer_.getPosition();
- }
-}
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.
+ */
+
+package org.apache.cassandra.io;
+
+import java.io.*;
+
+
+/**
+ * An implementation of the DataInputStream interface. This instance is completely thread
+ * unsafe.
+ *
+ * Author : Avinash Lakshman ( alakshman@facebook.com) & Prashant Malik ( pmalik@facebook.com )
+ */
+
+public final class DataInputBuffer extends DataInputStream
+{
+ private static class Buffer extends ByteArrayInputStream
+ {
+ public Buffer()
+ {
+ super(new byte[] {});
+ }
+
+ public void reset(byte[] input, int start, int length)
+ {
+ this.buf = input;
+ this.count = start + length;
+ this.mark = start;
+ this.pos = start;
+ }
+
+ public int getPosition()
+ {
+ return pos;
+ }
+
+ public void setPosition(int position)
+ {
+ pos = position;
+ }
+
+ public int getLength()
+ {
+ return count;
+ }
+ }
+
+ private Buffer buffer_;
+
+ /** Constructs a new empty buffer. */
+ public DataInputBuffer()
+ {
+ this(new Buffer());
+ }
+
+ private DataInputBuffer(Buffer buffer)
+ {
+ super(buffer);
+ this.buffer_ = buffer;
+ }
+
+ /** Resets the data that the buffer reads. */
+ public void reset(byte[] input, int length)
+ {
+ buffer_.reset(input, 0, length);
+ }
+
+ /** Resets the data that the buffer reads. */
+ public void reset(byte[] input, int start, int length)
+ {
+ buffer_.reset(input, start, length);
+ }
+
+ /** Returns the length of the input. */
+ public int getLength()
+ {
+ return buffer_.getLength();
+ }
+
+ public int getPosition()
+ {
+ return buffer_.getPosition();
+ }
+}
Modified: incubator/cassandra/trunk/src/java/org/apache/cassandra/io/DataOutputBuffer.java
URL: http://svn.apache.org/viewvc/incubator/cassandra/trunk/src/java/org/apache/cassandra/io/DataOutputBuffer.java?rev=799331&r1=799330&r2=799331&view=diff
==============================================================================
--- incubator/cassandra/trunk/src/java/org/apache/cassandra/io/DataOutputBuffer.java (original)
+++ incubator/cassandra/trunk/src/java/org/apache/cassandra/io/DataOutputBuffer.java Thu Jul 30 15:30:21 2009
@@ -1,108 +1,108 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you 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.
- */
-
-package org.apache.cassandra.io;
-
-import java.io.ByteArrayOutputStream;
-import java.io.DataInput;
-import java.io.DataOutputStream;
-import java.io.IOException;
-
-
-/**
- * An implementation of the DataOutputStream interface. This class is completely thread
- * unsafe.
- *
- * Author : Avinash Lakshman ( alakshman@facebook.com) & Prashant Malik ( pmalik@facebook.com )
- */
-public class DataOutputBuffer extends DataOutputStream
-{
- private static class Buffer extends ByteArrayOutputStream
- {
- public byte[] getData()
- {
- return buf;
- }
-
- public int getLength()
- {
- return count;
- }
-
- public void reset()
- {
- count = 0;
- }
-
- public void write(DataInput in, int len) throws IOException
- {
- int newcount = count + len;
- if (newcount > buf.length)
- {
- byte newbuf[] = new byte[Math.max(buf.length << 1, newcount)];
- System.arraycopy(buf, 0, newbuf, 0, count);
- buf = newbuf;
- }
- in.readFully(buf, count, len);
- count = newcount;
- }
- }
-
- private Buffer buffer;
-
- /** Constructs a new empty buffer. */
- public DataOutputBuffer()
- {
- this(new Buffer());
- }
-
- private DataOutputBuffer(Buffer buffer)
- {
- super(buffer);
- this.buffer = buffer;
- }
-
- /**
- * Returns the current contents of the buffer. Data is only valid to
- * {@link #getLength()}.
- */
- public byte[] getData()
- {
- return buffer.getData();
- }
-
- /** Returns the length of the valid data currently in the buffer. */
- public int getLength()
- {
- return buffer.getLength();
- }
-
- /** Resets the buffer to empty. */
- public DataOutputBuffer reset()
- {
- this.written = 0;
- buffer.reset();
- return this;
- }
-
- /** Writes bytes from a DataInput directly into the buffer. */
- public void write(DataInput in, int length) throws IOException
- {
- buffer.write(in, length);
- }
-}
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.
+ */
+
+package org.apache.cassandra.io;
+
+import java.io.ByteArrayOutputStream;
+import java.io.DataInput;
+import java.io.DataOutputStream;
+import java.io.IOException;
+
+
+/**
+ * An implementation of the DataOutputStream interface. This class is completely thread
+ * unsafe.
+ *
+ * Author : Avinash Lakshman ( alakshman@facebook.com) & Prashant Malik ( pmalik@facebook.com )
+ */
+public class DataOutputBuffer extends DataOutputStream
+{
+ private static class Buffer extends ByteArrayOutputStream
+ {
+ public byte[] getData()
+ {
+ return buf;
+ }
+
+ public int getLength()
+ {
+ return count;
+ }
+
+ public void reset()
+ {
+ count = 0;
+ }
+
+ public void write(DataInput in, int len) throws IOException
+ {
+ int newcount = count + len;
+ if (newcount > buf.length)
+ {
+ byte newbuf[] = new byte[Math.max(buf.length << 1, newcount)];
+ System.arraycopy(buf, 0, newbuf, 0, count);
+ buf = newbuf;
+ }
+ in.readFully(buf, count, len);
+ count = newcount;
+ }
+ }
+
+ private Buffer buffer;
+
+ /** Constructs a new empty buffer. */
+ public DataOutputBuffer()
+ {
+ this(new Buffer());
+ }
+
+ private DataOutputBuffer(Buffer buffer)
+ {
+ super(buffer);
+ this.buffer = buffer;
+ }
+
+ /**
+ * Returns the current contents of the buffer. Data is only valid to
+ * {@link #getLength()}.
+ */
+ public byte[] getData()
+ {
+ return buffer.getData();
+ }
+
+ /** Returns the length of the valid data currently in the buffer. */
+ public int getLength()
+ {
+ return buffer.getLength();
+ }
+
+ /** Resets the buffer to empty. */
+ public DataOutputBuffer reset()
+ {
+ this.written = 0;
+ buffer.reset();
+ return this;
+ }
+
+ /** Writes bytes from a DataInput directly into the buffer. */
+ public void write(DataInput in, int length) throws IOException
+ {
+ buffer.write(in, length);
+ }
+}