diff --git a/COPYING b/COPYING new file mode 100644 --- /dev/null +++ b/COPYING @@ -0,0 +1,502 @@ + GNU LESSER GENERAL PUBLIC LICENSE + Version 2.1, February 1999 + + Copyright (C) 1991, 1999 Free Software Foundation, Inc. + 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA + Everyone is permitted to copy and distribute verbatim copies + of this license document, but changing it is not allowed. + +[This is the first released version of the Lesser GPL. It also counts + as the successor of the GNU Library Public License, version 2, hence + the version number 2.1.] + + Preamble + + The licenses for most software are designed to take away your +freedom to share and change it. By contrast, the GNU General Public +Licenses are intended to guarantee your freedom to share and change +free software--to make sure the software is free for all its users. + + This license, the Lesser General Public License, applies to some +specially designated software packages--typically libraries--of the +Free Software Foundation and other authors who decide to use it. You +can use it too, but we suggest you first think carefully about whether +this license or the ordinary General Public License is the better +strategy to use in any particular case, based on the explanations below. + + When we speak of free software, we are referring to freedom of use, +not price. Our General Public Licenses are designed to make sure that +you have the freedom to distribute copies of free software (and charge +for this service if you wish); that you receive source code or can get +it if you want it; that you can change the software and use pieces of +it in new free programs; and that you are informed that you can do +these things. + + To protect your rights, we need to make restrictions that forbid +distributors to deny you these rights or to ask you to surrender these +rights. These restrictions translate to certain responsibilities for +you if you distribute copies of the library or if you modify it. + + For example, if you distribute copies of the library, whether gratis +or for a fee, you must give the recipients all the rights that we gave +you. You must make sure that they, too, receive or can get the source +code. If you link other code with the library, you must provide +complete object files to the recipients, so that they can relink them +with the library after making changes to the library and recompiling +it. And you must show them these terms so they know their rights. + + We protect your rights with a two-step method: (1) we copyright the +library, and (2) we offer you this license, which gives you legal +permission to copy, distribute and/or modify the library. + + To protect each distributor, we want to make it very clear that +there is no warranty for the free library. Also, if the library is +modified by someone else and passed on, the recipients should know +that what they have is not the original version, so that the original +author's reputation will not be affected by problems that might be +introduced by others. + + Finally, software patents pose a constant threat to the existence of +any free program. We wish to make sure that a company cannot +effectively restrict the users of a free program by obtaining a +restrictive license from a patent holder. Therefore, we insist that +any patent license obtained for a version of the library must be +consistent with the full freedom of use specified in this license. + + Most GNU software, including some libraries, is covered by the +ordinary GNU General Public License. This license, the GNU Lesser +General Public License, applies to certain designated libraries, and +is quite different from the ordinary General Public License. We use +this license for certain libraries in order to permit linking those +libraries into non-free programs. + + When a program is linked with a library, whether statically or using +a shared library, the combination of the two is legally speaking a +combined work, a derivative of the original library. The ordinary +General Public License therefore permits such linking only if the +entire combination fits its criteria of freedom. The Lesser General +Public License permits more lax criteria for linking other code with +the library. + + We call this license the "Lesser" General Public License because it +does Less to protect the user's freedom than the ordinary General +Public License. It also provides other free software developers Less +of an advantage over competing non-free programs. These disadvantages +are the reason we use the ordinary General Public License for many +libraries. However, the Lesser license provides advantages in certain +special circumstances. + + For example, on rare occasions, there may be a special need to +encourage the widest possible use of a certain library, so that it becomes +a de-facto standard. To achieve this, non-free programs must be +allowed to use the library. A more frequent case is that a free +library does the same job as widely used non-free libraries. In this +case, there is little to gain by limiting the free library to free +software only, so we use the Lesser General Public License. + + In other cases, permission to use a particular library in non-free +programs enables a greater number of people to use a large body of +free software. For example, permission to use the GNU C Library in +non-free programs enables many more people to use the whole GNU +operating system, as well as its variant, the GNU/Linux operating +system. + + Although the Lesser General Public License is Less protective of the +users' freedom, it does ensure that the user of a program that is +linked with the Library has the freedom and the wherewithal to run +that program using a modified version of the Library. + + The precise terms and conditions for copying, distribution and +modification follow. Pay close attention to the difference between a +"work based on the library" and a "work that uses the library". The +former contains code derived from the library, whereas the latter must +be combined with the library in order to run. + + GNU LESSER GENERAL PUBLIC LICENSE + TERMS AND CONDITIONS FOR COPYING, DISTRIBUTION AND MODIFICATION + + 0. This License Agreement applies to any software library or other +program which contains a notice placed by the copyright holder or +other authorized party saying it may be distributed under the terms of +this Lesser General Public License (also called "this License"). +Each licensee is addressed as "you". + + A "library" means a collection of software functions and/or data +prepared so as to be conveniently linked with application programs +(which use some of those functions and data) to form executables. + + The "Library", below, refers to any such software library or work +which has been distributed under these terms. A "work based on the +Library" means either the Library or any derivative work under +copyright law: that is to say, a work containing the Library or a +portion of it, either verbatim or with modifications and/or translated +straightforwardly into another language. (Hereinafter, translation is +included without limitation in the term "modification".) + + "Source code" for a work means the preferred form of the work for +making modifications to it. For a library, complete source code means +all the source code for all modules it contains, plus any associated +interface definition files, plus the scripts used to control compilation +and installation of the library. + + Activities other than copying, distribution and modification are not +covered by this License; they are outside its scope. The act of +running a program using the Library is not restricted, and output from +such a program is covered only if its contents constitute a work based +on the Library (independent of the use of the Library in a tool for +writing it). Whether that is true depends on what the Library does +and what the program that uses the Library does. + + 1. You may copy and distribute verbatim copies of the Library's +complete source code as you receive it, in any medium, provided that +you conspicuously and appropriately publish on each copy an +appropriate copyright notice and disclaimer of warranty; keep intact +all the notices that refer to this License and to the absence of any +warranty; and distribute a copy of this License along with the +Library. + + You may charge a fee for the physical act of transferring a copy, +and you may at your option offer warranty protection in exchange for a +fee. + + 2. You may modify your copy or copies of the Library or any portion +of it, thus forming a work based on the Library, and copy and +distribute such modifications or work under the terms of Section 1 +above, provided that you also meet all of these conditions: + + a) The modified work must itself be a software library. + + b) You must cause the files modified to carry prominent notices + stating that you changed the files and the date of any change. + + c) You must cause the whole of the work to be licensed at no + charge to all third parties under the terms of this License. + + d) If a facility in the modified Library refers to a function or a + table of data to be supplied by an application program that uses + the facility, other than as an argument passed when the facility + is invoked, then you must make a good faith effort to ensure that, + in the event an application does not supply such function or + table, the facility still operates, and performs whatever part of + its purpose remains meaningful. + + (For example, a function in a library to compute square roots has + a purpose that is entirely well-defined independent of the + application. Therefore, Subsection 2d requires that any + application-supplied function or table used by this function must + be optional: if the application does not supply it, the square + root function must still compute square roots.) + +These requirements apply to the modified work as a whole. If +identifiable sections of that work are not derived from the Library, +and can be reasonably considered independent and separate works in +themselves, then this License, and its terms, do not apply to those +sections when you distribute them as separate works. But when you +distribute the same sections as part of a whole which is a work based +on the Library, the distribution of the whole must be on the terms of +this License, whose permissions for other licensees extend to the +entire whole, and thus to each and every part regardless of who wrote +it. + +Thus, it is not the intent of this section to claim rights or contest +your rights to work written entirely by you; rather, the intent is to +exercise the right to control the distribution of derivative or +collective works based on the Library. + +In addition, mere aggregation of another work not based on the Library +with the Library (or with a work based on the Library) on a volume of +a storage or distribution medium does not bring the other work under +the scope of this License. + + 3. You may opt to apply the terms of the ordinary GNU General Public +License instead of this License to a given copy of the Library. To do +this, you must alter all the notices that refer to this License, so +that they refer to the ordinary GNU General Public License, version 2, +instead of to this License. (If a newer version than version 2 of the +ordinary GNU General Public License has appeared, then you can specify +that version instead if you wish.) Do not make any other change in +these notices. + + Once this change is made in a given copy, it is irreversible for +that copy, so the ordinary GNU General Public License applies to all +subsequent copies and derivative works made from that copy. + + This option is useful when you wish to copy part of the code of +the Library into a program that is not a library. + + 4. You may copy and distribute the Library (or a portion or +derivative of it, under Section 2) in object code or executable form +under the terms of Sections 1 and 2 above provided that you accompany +it with the complete corresponding machine-readable source code, which +must be distributed under the terms of Sections 1 and 2 above on a +medium customarily used for software interchange. + + If distribution of object code is made by offering access to copy +from a designated place, then offering equivalent access to copy the +source code from the same place satisfies the requirement to +distribute the source code, even though third parties are not +compelled to copy the source along with the object code. + + 5. A program that contains no derivative of any portion of the +Library, but is designed to work with the Library by being compiled or +linked with it, is called a "work that uses the Library". Such a +work, in isolation, is not a derivative work of the Library, and +therefore falls outside the scope of this License. + + However, linking a "work that uses the Library" with the Library +creates an executable that is a derivative of the Library (because it +contains portions of the Library), rather than a "work that uses the +library". The executable is therefore covered by this License. +Section 6 states terms for distribution of such executables. + + When a "work that uses the Library" uses material from a header file +that is part of the Library, the object code for the work may be a +derivative work of the Library even though the source code is not. +Whether this is true is especially significant if the work can be +linked without the Library, or if the work is itself a library. The +threshold for this to be true is not precisely defined by law. + + If such an object file uses only numerical parameters, data +structure layouts and accessors, and small macros and small inline +functions (ten lines or less in length), then the use of the object +file is unrestricted, regardless of whether it is legally a derivative +work. (Executables containing this object code plus portions of the +Library will still fall under Section 6.) + + Otherwise, if the work is a derivative of the Library, you may +distribute the object code for the work under the terms of Section 6. +Any executables containing that work also fall under Section 6, +whether or not they are linked directly with the Library itself. + + 6. As an exception to the Sections above, you may also combine or +link a "work that uses the Library" with the Library to produce a +work containing portions of the Library, and distribute that work +under terms of your choice, provided that the terms permit +modification of the work for the customer's own use and reverse +engineering for debugging such modifications. + + You must give prominent notice with each copy of the work that the +Library is used in it and that the Library and its use are covered by +this License. You must supply a copy of this License. If the work +during execution displays copyright notices, you must include the +copyright notice for the Library among them, as well as a reference +directing the user to the copy of this License. Also, you must do one +of these things: + + a) Accompany the work with the complete corresponding + machine-readable source code for the Library including whatever + changes were used in the work (which must be distributed under + Sections 1 and 2 above); and, if the work is an executable linked + with the Library, with the complete machine-readable "work that + uses the Library", as object code and/or source code, so that the + user can modify the Library and then relink to produce a modified + executable containing the modified Library. (It is understood + that the user who changes the contents of definitions files in the + Library will not necessarily be able to recompile the application + to use the modified definitions.) + + b) Use a suitable shared library mechanism for linking with the + Library. A suitable mechanism is one that (1) uses at run time a + copy of the library already present on the user's computer system, + rather than copying library functions into the executable, and (2) + will operate properly with a modified version of the library, if + the user installs one, as long as the modified version is + interface-compatible with the version that the work was made with. + + c) Accompany the work with a written offer, valid for at + least three years, to give the same user the materials + specified in Subsection 6a, above, for a charge no more + than the cost of performing this distribution. + + d) If distribution of the work is made by offering access to copy + from a designated place, offer equivalent access to copy the above + specified materials from the same place. + + e) Verify that the user has already received a copy of these + materials or that you have already sent this user a copy. + + For an executable, the required form of the "work that uses the +Library" must include any data and utility programs needed for +reproducing the executable from it. However, as a special exception, +the materials to be distributed need not include anything that is +normally distributed (in either source or binary form) with the major +components (compiler, kernel, and so on) of the operating system on +which the executable runs, unless that component itself accompanies +the executable. + + It may happen that this requirement contradicts the license +restrictions of other proprietary libraries that do not normally +accompany the operating system. Such a contradiction means you cannot +use both them and the Library together in an executable that you +distribute. + + 7. You may place library facilities that are a work based on the +Library side-by-side in a single library together with other library +facilities not covered by this License, and distribute such a combined +library, provided that the separate distribution of the work based on +the Library and of the other library facilities is otherwise +permitted, and provided that you do these two things: + + a) Accompany the combined library with a copy of the same work + based on the Library, uncombined with any other library + facilities. This must be distributed under the terms of the + Sections above. + + b) Give prominent notice with the combined library of the fact + that part of it is a work based on the Library, and explaining + where to find the accompanying uncombined form of the same work. + + 8. You may not copy, modify, sublicense, link with, or distribute +the Library except as expressly provided under this License. Any +attempt otherwise to copy, modify, sublicense, link with, or +distribute the Library is void, and will automatically terminate your +rights under this License. However, parties who have received copies, +or rights, from you under this License will not have their licenses +terminated so long as such parties remain in full compliance. + + 9. You are not required to accept this License, since you have not +signed it. However, nothing else grants you permission to modify or +distribute the Library or its derivative works. These actions are +prohibited by law if you do not accept this License. Therefore, by +modifying or distributing the Library (or any work based on the +Library), you indicate your acceptance of this License to do so, and +all its terms and conditions for copying, distributing or modifying +the Library or works based on it. + + 10. Each time you redistribute the Library (or any work based on the +Library), the recipient automatically receives a license from the +original licensor to copy, distribute, link with or modify the Library +subject to these terms and conditions. You may not impose any further +restrictions on the recipients' exercise of the rights granted herein. +You are not responsible for enforcing compliance by third parties with +this License. + + 11. If, as a consequence of a court judgment or allegation of patent +infringement or for any other reason (not limited to patent issues), +conditions are imposed on you (whether by court order, agreement or +otherwise) that contradict the conditions of this License, they do not +excuse you from the conditions of this License. If you cannot +distribute so as to satisfy simultaneously your obligations under this +License and any other pertinent obligations, then as a consequence you +may not distribute the Library at all. For example, if a patent +license would not permit royalty-free redistribution of the Library by +all those who receive copies directly or indirectly through you, then +the only way you could satisfy both it and this License would be to +refrain entirely from distribution of the Library. + +If any portion of this section is held invalid or unenforceable under any +particular circumstance, the balance of the section is intended to apply, +and the section as a whole is intended to apply in other circumstances. + +It is not the purpose of this section to induce you to infringe any +patents or other property right claims or to contest validity of any +such claims; this section has the sole purpose of protecting the +integrity of the free software distribution system which is +implemented by public license practices. Many people have made +generous contributions to the wide range of software distributed +through that system in reliance on consistent application of that +system; it is up to the author/donor to decide if he or she is willing +to distribute software through any other system and a licensee cannot +impose that choice. + +This section is intended to make thoroughly clear what is believed to +be a consequence of the rest of this License. + + 12. If the distribution and/or use of the Library is restricted in +certain countries either by patents or by copyrighted interfaces, the +original copyright holder who places the Library under this License may add +an explicit geographical distribution limitation excluding those countries, +so that distribution is permitted only in or among countries not thus +excluded. In such case, this License incorporates the limitation as if +written in the body of this License. + + 13. The Free Software Foundation may publish revised and/or new +versions of the Lesser General Public License from time to time. +Such new versions will be similar in spirit to the present version, +but may differ in detail to address new problems or concerns. + +Each version is given a distinguishing version number. If the Library +specifies a version number of this License which applies to it and +"any later version", you have the option of following the terms and +conditions either of that version or of any later version published by +the Free Software Foundation. If the Library does not specify a +license version number, you may choose any version ever published by +the Free Software Foundation. + + 14. If you wish to incorporate parts of the Library into other free +programs whose distribution conditions are incompatible with these, +write to the author to ask for permission. For software which is +copyrighted by the Free Software Foundation, write to the Free +Software Foundation; we sometimes make exceptions for this. Our +decision will be guided by the two goals of preserving the free status +of all derivatives of our free software and of promoting the sharing +and reuse of software generally. + + NO WARRANTY + + 15. BECAUSE THE LIBRARY IS LICENSED FREE OF CHARGE, THERE IS NO +WARRANTY FOR THE LIBRARY, TO THE EXTENT PERMITTED BY APPLICABLE LAW. +EXCEPT WHEN OTHERWISE STATED IN WRITING THE COPYRIGHT HOLDERS AND/OR +OTHER PARTIES PROVIDE THE LIBRARY "AS IS" WITHOUT WARRANTY OF ANY +KIND, EITHER EXPRESSED OR IMPLIED, INCLUDING, BUT NOT LIMITED TO, THE +IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR +PURPOSE. THE ENTIRE RISK AS TO THE QUALITY AND PERFORMANCE OF THE +LIBRARY IS WITH YOU. SHOULD THE LIBRARY PROVE DEFECTIVE, YOU ASSUME +THE COST OF ALL NECESSARY SERVICING, REPAIR OR CORRECTION. + + 16. IN NO EVENT UNLESS REQUIRED BY APPLICABLE LAW OR AGREED TO IN +WRITING WILL ANY COPYRIGHT HOLDER, OR ANY OTHER PARTY WHO MAY MODIFY +AND/OR REDISTRIBUTE THE LIBRARY AS PERMITTED ABOVE, BE LIABLE TO YOU +FOR DAMAGES, INCLUDING ANY GENERAL, SPECIAL, INCIDENTAL OR +CONSEQUENTIAL DAMAGES ARISING OUT OF THE USE OR INABILITY TO USE THE +LIBRARY (INCLUDING BUT NOT LIMITED TO LOSS OF DATA OR DATA BEING +RENDERED INACCURATE OR LOSSES SUSTAINED BY YOU OR THIRD PARTIES OR A +FAILURE OF THE LIBRARY TO OPERATE WITH ANY OTHER SOFTWARE), EVEN IF +SUCH HOLDER OR OTHER PARTY HAS BEEN ADVISED OF THE POSSIBILITY OF SUCH +DAMAGES. + + END OF TERMS AND CONDITIONS + + How to Apply These Terms to Your New Libraries + + If you develop a new library, and you want it to be of the greatest +possible use to the public, we recommend making it free software that +everyone can redistribute and change. You can do so by permitting +redistribution under these terms (or, alternatively, under the terms of the +ordinary General Public License). + + To apply these terms, attach the following notices to the library. It is +safest to attach them to the start of each source file to most effectively +convey the exclusion of warranty; and each file should have at least the +"copyright" line and a pointer to where the full notice is found. + + + Copyright (C) + + This library is free software; you can redistribute it and/or + modify it under the terms of the GNU Lesser General Public + License as published by the Free Software Foundation; either + version 2.1 of the License, or (at your option) any later version. + + This library is distributed in the hope that it will be useful, + but WITHOUT ANY WARRANTY; without even the implied warranty of + MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + Lesser General Public License for more details. + + You should have received a copy of the GNU Lesser General Public + License along with this library; if not, write to the Free Software + Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA + +Also add information on how to contact you by electronic and paper mail. + +You should also get your employer (if you work as a programmer) or your +school, if any, to sign a "copyright disclaimer" for the library, if +necessary. Here is a sample; alter the names: + + Yoyodyne, Inc., hereby disclaims all copyright interest in the + library `Frob' (a library for tweaking knobs) written by James Random Hacker. + + , 1 April 1990 + Ty Coon, President of Vice + +That's all there is to it! diff --git a/COPYRIGHT b/COPYRIGHT new file mode 100644 --- /dev/null +++ b/COPYRIGHT @@ -0,0 +1,18 @@ +Copyright (C) 2009-2010 Andrejs Sisojevs + + This library is free software; you can redistribute it and/or + modify it under the terms of the GNU Lesser General Public + License as published by the Free Software Foundation; either + version 2.1 of the License, or (at your option) any later version. + + This library is distributed in the hope that it will be useful, + but WITHOUT ANY WARRANTY; without even the implied warranty of + MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + Lesser General Public License for more details. + + You should have received a copy of the GNU Lesser General Public + License along with this library; if not, write to the Free Software + Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA + +Please see COPYING for the license. + diff --git a/Control/Concurrent/ConcurrentUISupport.hs b/Control/Concurrent/ConcurrentUISupport.hs new file mode 100644 --- /dev/null +++ b/Control/Concurrent/ConcurrentUISupport.hs @@ -0,0 +1,227 @@ +{- +Copyright (C) 2009-2010 Andrejs Sisojevs + +All rights reserved. + +For license and copyright information, see the file COPYRIGHT +-} + +-------------------------------------------------------------------------- +-------------------------------------------------------------------------- + +-- | A bottleneck for UI from multiple threads. Prevents mixing of output +-- messages. Also makes possible a primitive form of input. By mastering +-- @'cuisUIInput_E'@ and @'cuisUIOutput_E'@ it's possible +-- to construct @'ConcurrentUISupport'@ with different than UI I/O. +module Control.Concurrent.ConcurrentUISupport where + +import Control.Concurrent +import Control.Concurrent.MVar +import Control.Concurrent.STM +import Control.Monad +import qualified Control.Exception as E +import Data.List +import qualified Data.Map as M +import Data.Map (Map) +import Data.MyHelpers +import Data.Typeable +import System.IO + +-- * ConcurrentUISupport administration + +type AskerID = String +type ReadFromConsoleStr = String +type ThreadFinisher = IO () + +data CUISCommand = + NewPage_CUISC + | PlusLine_CUISC + | PutStrLn_CUISC String + | ReadStr_CUISC AskerID deriving (Eq) + +data ConcurrentUISupport = + ConcurrentUISupport { + -- | Supported threads are welcome to write their commands here. + -- It is recommended not to use this channel directly - + -- use @'cuisWrite'@, @'cuisReadLn'@ and other mediators instead. + cuisInputTChan :: TChan [CUISCommand] + -- | Map of feedback chans leading to threads, that issue the @'ReadStr_CUISC'@ commands. + , cuisOutputChans :: TVar (Map AskerID (TChan ReadFromConsoleStr)) + -- | Call this, when you want to finish supporter thread. + , cuisFinish :: ThreadFinisher + -- | Was @'cuisFinish'@ called? + , cuisSupporterStopCheck :: STM Bool + -- | What to do, when supporter thread finishes? + , cuisFinalizer :: IO () + -- | How to input, when @'ReadStr_CUISC'@ command comes? + , cuisUIInput_E :: Either (STM String) (IO String) + -- | How to output, when @'PutStrLn_CUISC'@ command comes? + , cuisUIOutput_E :: Either (String -> STM ()) (String -> IO ()) + -- | If set, then call it on each @'NewPage_CUISC'@ command. + , cuisPressAnyKey_uiCmd :: Maybe (IO ()) + -- | For @(Just i)@, if count of printed lines is > @i@, + -- then automatically issue @'NewPage_CUISC'@ command and + -- set lines counter to 0. + , cuisPressAnyKey_linesCountCond :: Maybe Int + } + +defaultConcurrentUISupportIO :: IO ConcurrentUISupport +defaultConcurrentUISupportIO = do + input_tch <- atomically newTChan + breaker_tv <- atomically $ newTVar False + out_chans <- atomically $ newTVar (M.empty) + return ConcurrentUISupport { + cuisInputTChan = input_tch + , cuisOutputChans = out_chans + , cuisFinish = atomically $ writeTVar breaker_tv True + , cuisFinalizer = return () + , cuisSupporterStopCheck = readTVar breaker_tv + , cuisUIInput_E = Right getLine + , cuisUIOutput_E = Right putStrLn + , cuisPressAnyKey_uiCmd = Just pressAnyKey + , cuisPressAnyKey_linesCountCond = Just 22 + } + +-- | UI in console + log to file. +defaultConcurrentUISupportIO_logInFile :: FilePath -> IOMode -> IO ConcurrentUISupport +defaultConcurrentUISupportIO_logInFile f_name io_mode = do + input_tch <- atomically newTChan + breaker_tv <- atomically $ newTVar False + out_chans <- atomically $ newTVar (M.empty) + h <- openFile f_name io_mode + hSetBuffering h LineBuffering + return ConcurrentUISupport { + cuisInputTChan = input_tch + , cuisOutputChans = out_chans + , cuisFinish = atomically $ writeTVar breaker_tv True + , cuisFinalizer = hClose h + , cuisSupporterStopCheck = readTVar breaker_tv + , cuisUIInput_E = Right (do { s <- getLine; hPutStrLn h s; return s}) + , cuisUIOutput_E = Right (\ s -> (hPutStrLn h s >> putStrLn s)) + , cuisPressAnyKey_uiCmd = Just pressAnyKey + , cuisPressAnyKey_linesCountCond = Just 22 + } + +-- | Whatever @'cuisUIInput_E'@ is - @Left@ or @Right@ - flattern it into IO. +cuisUIInput :: ConcurrentUISupport -> IO String +cuisUIInput cuis = + case cuisUIInput_E cuis of + Left stm_a -> atomically stm_a + Right io_a -> io_a + +-- | Whatever @'cuisUIOutput_E'@ is - @Left@ or @Right@ - flattern it into IO. +cuisUIOutput :: ConcurrentUISupport -> String -> IO () +cuisUIOutput cuis s = + case cuisUIOutput_E cuis of + Left stm_a -> atomically $ stm_a s + Right io_a -> io_a s + +-- | Not to be confised with @'cuisUIInput'@, which interfaces with user. +-- This /input/ interfaces with threads, that makes use of it. +-- +-- A wrapper around @'cuisInputTChan'@. +cuisInput :: ConcurrentUISupport -> [CUISCommand] -> IO () +cuisInput cuis cmds = atomically $ writeTChan (cuisInputTChan cuis) cmds + +acquirePutStr_fromCmdsList :: [CUISCommand] -> Maybe CUISCommand +acquirePutStr_fromCmdsList cmds = + let find_p e = + case e of + PutStrLn_CUISC _ -> True + _ -> False + in find find_p cmds +acquireReadStr_fromCmdsList :: [CUISCommand] -> Maybe CUISCommand +acquireReadStr_fromCmdsList cmds = + let find_p e = + case e of + ReadStr_CUISC _ -> True + _ -> False + in find find_p cmds + +runConcurrentUISupport :: ConcurrentUISupport -> IO () +runConcurrentUISupport cuis = do + _ <- forkIO $ + let _cycle :: Int -> IO () + _cycle i = do + (stop, mb_cmds) <- atomically $ do + stop <- cuisSupporterStopCheck cuis + case stop of + True -> return (stop, Nothing) + False -> do + cmds <- readTChan $ cuisInputTChan cuis + return (stop, Just cmds) + new_i <- case mb_cmds of + Nothing -> return i + Just cmds -> do + -- Process PutStrLn_CUISC + case acquirePutStr_fromCmdsList cmds of + Nothing -> return () + Just (PutStrLn_CUISC s) -> cuisUIOutput cuis s + -- Process ReadStr_CUISC + case acquireReadStr_fromCmdsList cmds of + Nothing -> return () + Just (ReadStr_CUISC asker_id) -> do + t <- cuisUIInput cuis + atomically $ do + outs_map <- readTVar (cuisOutputChans cuis) + asker_chan <- case M.lookup asker_id outs_map of + Nothing -> do + asker_chan <- newTChan + modifyTVar_ (cuisOutputChans cuis) (M.insert asker_id asker_chan) + return asker_chan + Just asker_chan -> return asker_chan + writeTChan asker_chan t + -- Process NewPage_CUISC, PlusLine_CUISC + let cond_1 = (cuisPressAnyKey_linesCountCond cuis >>= \ bnd -> return (i >= bnd) ) == Just True + let cond_2 = NewPage_CUISC `elem` cmds + case cond_1 || cond_2 of + True -> return 0 << whenJust (cuisPressAnyKey_uiCmd cuis) + False -> return (i + (bool2num $ PlusLine_CUISC `elem` cmds)) + case stop of + True -> return () -- exit + False -> _cycle new_i + in E.finally (_cycle 0) (cuisFinalizer cuis) + ------------------- + return () + +-- * Use these functions to deal with ConcurrentUISupport + +cuisWriteSTM :: ConcurrentUISupport -> String -> STM () +cuisWriteSTM cuis s = + mapM_ (\ _s -> writeTChan (cuisInputTChan cuis) [PlusLine_CUISC, PutStrLn_CUISC _s]) (lines s) + +cuisWrite :: ConcurrentUISupport -> String -> IO () +cuisWrite cuis s = + mapM_ (\ _s -> cuisInput cuis [PlusLine_CUISC, PutStrLn_CUISC _s]) (lines s) + +cuisNewPage :: ConcurrentUISupport -> IO () +cuisNewPage cuis = + cuisInput cuis [NewPage_CUISC] + +cuisReadLn :: ConcurrentUISupport -> AskerID -> IO ReadFromConsoleStr +cuisReadLn cuis asker_id = do + cuisInput cuis [ReadStr_CUISC asker_id, PlusLine_CUISC] + atomically $ do + outs_map <- readTVar $ cuisOutputChans cuis + case M.lookup asker_id outs_map of + Nothing -> retry + Just asker_chan -> readTChan asker_chan + +------------------------------------------------------------ +-- * Few helpers for dealing with exceptions + +reportExceptionIfAny :: (String -> IO ()) -> String -> IO a -> IO a +reportExceptionIfAny reportStr caller_f_name = E.handle (\ se@(E.SomeException e) -> reportStr ("An error occurred in function '" ++ caller_f_name ++ "'. Type: " ++ (show $ typeRepTyCon $ typeOf e) ++ ". Representation: " ++ show se) >> E.throw (se :: E.SomeException)) + +printExceptionIfAny :: String -> IO a -> IO a +printExceptionIfAny = reportExceptionIfAny putStrLn + +reportExceptionIfAny_2 :: ConcurrentUISupport -> String -> IO a -> IO a +reportExceptionIfAny_2 cuis = reportExceptionIfAny (cuisWrite cuis) + +reportExceptionIfAnySTM :: (String -> STM ()) -> String -> STM a -> STM a +reportExceptionIfAnySTM reportStr caller_f_name stma = catchSTM stma (\ se@(E.SomeException e) -> reportStr ("An error occurred in function '" ++ caller_f_name ++ "'. Type: " ++ (show $ typeRepTyCon $ typeOf e) ++ ". Representation: " ++ show se) >> E.throw (se :: E.SomeException)) + +reportExceptionIfAnySTM_2 :: ConcurrentUISupport -> String -> STM a -> STM a +reportExceptionIfAnySTM_2 cuis = reportExceptionIfAnySTM (cuisWriteSTM cuis) + diff --git a/Control/Concurrent/PriorityChansConverger.hs b/Control/Concurrent/PriorityChansConverger.hs new file mode 100644 --- /dev/null +++ b/Control/Concurrent/PriorityChansConverger.hs @@ -0,0 +1,20 @@ +{- +Copyright (C) 2009-2010 Andrejs Sisojevs + +All rights reserved. + +For license and copyright information, see the file COPYRIGHT +-} + +-------------------------------------------------------------------------- +-------------------------------------------------------------------------- + +module Control.Concurrent.PriorityChansConverger ( + module Control.Concurrent.PriorityChansConverger.PriorityChansConverger + , module Control.Concurrent.PriorityChansConverger.Commons + , module Control.Concurrent.STM.TChan.ExtsCommons + ) where + +import Control.Concurrent.PriorityChansConverger.Commons +import Control.Concurrent.PriorityChansConverger.PriorityChansConverger +import Control.Concurrent.STM.TChan.ExtsCommons diff --git a/Control/Concurrent/PriorityChansConverger/Commons.hs b/Control/Concurrent/PriorityChansConverger/Commons.hs new file mode 100644 --- /dev/null +++ b/Control/Concurrent/PriorityChansConverger/Commons.hs @@ -0,0 +1,26 @@ +{- +Copyright (C) 2009-2010 Andrejs Sisojevs + +All rights reserved. + +For license and copyright information, see the file COPYRIGHT +-} + +-------------------------------------------------------------------------- +-------------------------------------------------------------------------- + +module Control.Concurrent.PriorityChansConverger.Commons where + +import Control.Concurrent.STM.TChan.ExtsCommons + +type StartPriority = ChanContentAmountMeasure +type CurrentPriority = StartPriority +type PCC_ContainsGivenSpecification = Bool +type NothingElseIsInStruct = Bool +type InterruptShouldWe = Bool + +data FailureReasonWPCC = BadKey_FRWPCC | ChanFull_FRWPCC +instance Show FailureReasonWPCC where + show fr = case fr of + BadKey_FRWPCC -> "Error in adressing priority channel in converger: bad key." + ChanFull_FRWPCC -> "There's no free space in priority channel - it's full." diff --git a/Control/Concurrent/PriorityChansConverger/PriorityChansConverger.hs b/Control/Concurrent/PriorityChansConverger/PriorityChansConverger.hs new file mode 100644 --- /dev/null +++ b/Control/Concurrent/PriorityChansConverger/PriorityChansConverger.hs @@ -0,0 +1,284 @@ +{- +Copyright (C) 2009-2010 Andrejs Sisojevs + +All rights reserved. + +For license and copyright information, see the file COPYRIGHT +-} + +-------------------------------------------------------------------------- +-------------------------------------------------------------------------- + +-- | This module wraps STM PCC with IO. +-- Some locks are added to balance heavyweight transactions, to reduce \"waste +-- efforts\". +module Control.Concurrent.PriorityChansConverger.PriorityChansConverger where + +import Control.Concurrent +import Control.Concurrent.MVar +import Control.Concurrent.PriorityChansConverger.Commons +import Control.Concurrent.PriorityChansConverger.PriorityChansConvergerSTM +import Control.Concurrent.ConcurrentUISupport +import Control.Concurrent.STM +import Control.Concurrent.STM.TChan.ExtsCommons +import Control.Exception +import Control.Monad +import Data.IORef +import Data.List +import qualified Data.Map as M +import Data.Map (Map) +import Data.MyHelpers +import System.IO + +__locksObserverEnabled = False +---------------------------------------- + +-- * PriorityChansConverger ADT and it's administration + +data PCC_Lock = PCC_Lock | PCC_IOInterruptableRead | PCC_IOInterruptableWrite deriving (Eq, Ord, Show) + +-- | Wrapper around @'PriorityChansConvergerSTM'@. Added locks. @'Show'@ instance +-- is defined using @'unsafePerformIO'@. +data Ord k => PriorityChansConverger k e = PriorityChansConverger { + pccSTM :: PriorityChansConvergerSTM k e + , pccLocks :: Map PCC_Lock (MVar ()) +} + +reportIfPossible :: Ord k => PriorityChansConverger k e -> String -> IO () +reportIfPossible pcc s = + case pccConcurrentUISupport $ pccSTM pcc of + Nothing -> return () + Just cuis -> cuisWrite cuis s + +maybeObserveExceptions :: Ord k => PriorityChansConverger k e -> FunctionName -> IO a -> IO a +maybeObserveExceptions pcc f_name io_a = + case pccConcurrentUISupport $ pccSTM pcc of + Nothing -> io_a + Just cuis -> reportExceptionIfAny_2 cuis f_name io_a + +-- | Wrapper around @'newPriorityChansConverger_wCUIS'@ constructor. +-- +-- @newPriorityChansConverger = 'newPriorityChansConverger_wCUIS' Nothing@ +newPriorityChansConverger :: Ord k => Map k (StartPriority, ChanCapacity) -> IO (PriorityChansConverger k e) +newPriorityChansConverger = newPriorityChansConverger_wCUIS Nothing + +-- | Wrapper around @'newPriorityChansConvergerSTM_wCUIS'@ constructor. +-- With an option to enable utility @'ConcurrentUISupport'@ +-- (which is useful for debugging) +newPriorityChansConverger_wCUIS :: Ord k => Maybe ConcurrentUISupport -> Map k (StartPriority, ChanCapacity) -> IO (PriorityChansConverger k e) +newPriorityChansConverger_wCUIS mb_cuis chan_priors_map = do + pcc <- let constr = atomically $ newPriorityChansConvergerSTM_wCUIS mb_cuis chan_priors_map + in case mb_cuis of + Nothing -> constr + Just cuis -> reportExceptionIfAny_2 cuis f_name constr + l <- newMVar () + return PriorityChansConverger { + pccSTM = pcc + , pccLocks = M.fromList [(PCC_Lock, l)] + } + where f_name = "newPriorityChansConverger_wCUIS" + +type FunctionName = String +type LockUserName = FunctionName +-- | Sort the locks and lock them. +-- +-- WARNING !!! Asynchronous exceptions may cause loss of locks! +lockPCC :: Ord k => LockUserName -> [PCC_Lock] -> PriorityChansConverger k e -> IO () +lockPCC locks_user_f_name locks pcc = + mapM_ + (\ l -> + case M.lookup l (pccLocks pcc) of + Nothing -> error ("An error occurred in 'lockPCC': no lock for key '" ++ show l ++ "' exists.") + Just l_mv -> takeMVar l_mv >> when __locksObserverEnabled (reportIfPossible pcc ("Lock '" ++ show l ++ "' taken by function '" ++ locks_user_f_name ++ "'.")) + ) + (map head $ group $ sort locks) + +-- | Reverse sort the locks and unlock them. +-- +-- WARNING !!! Asynchronous exceptions may cause loss of locks! +unlockPCC :: Ord k => LockUserName -> [PCC_Lock] -> PriorityChansConverger k e -> IO () +unlockPCC locks_user_f_name locks pcc = + mapM_ + (\ l -> + case M.lookup l (pccLocks pcc) of + Nothing -> error ("An error occurred in 'unlockPCC': no lock for key '" ++ show l ++ "' exists.") + Just l_mv -> do + s <- tryPutMVar l_mv () + when __locksObserverEnabled $ reportIfPossible pcc $ case s of + True -> "Lock '" ++ show l ++ "' released by function '" ++ locks_user_f_name ++ "'." + False -> "Function '" ++ locks_user_f_name ++ "' wanted to release the lock '" ++ show l ++ "', but it appered to be already released." + ) + (map head $ group $ reverse $ sort locks) + +-- | Locking IO action wrapper. +-- +-- WARNING !!! Asynchronous exceptions may cause loss of locks! +withLockedDo :: Ord k => LockUserName -> [PCC_Lock] -> PriorityChansConverger k e -> IO a -> IO a +withLockedDo locks_user_f_name locks pcc act = block $ do + lockPCC locks_user_f_name locks pcc + a <- finally (unblock act) (unlockPCC locks_user_f_name locks pcc) + return a + +--------------------------------------------------------------------------------------------------------------------------- +-- * PriorityChansConverger mutation +--------------------------------------------------------------------------------------------------------------------------- + +-- | Wrapper around @'isOfStructPCC_STM'@. Used locks: none. +isOfStructPCC :: Ord k => Map k (StartPriority, ChanCapacity) -> PriorityChansConverger k e -> IO (PCC_ContainsGivenSpecification, NothingElseIsInStruct) +isOfStructPCC mp pcc = maybeObserveExceptions pcc f_name $ atomically $ isOfStructPCC_STM mp (pccSTM pcc) + where f_name = "isOfStructPCC" + +-- | Wrapper around @'mutatePCC_STM'@. Used locks: @['PCC_Lock']@ +mutatePCC :: Ord k => Map k (StartPriority, ChanCapacity) -> PriorityChansConverger k e -> IO (PriorityChansConverger k e) +mutatePCC chan_priors_map pcc = withLockedDo f_name [PCC_Lock] pcc $ do + new_pcc_stm <- maybeObserveExceptions pcc f_name $ atomically $ mutatePCC_STM chan_priors_map (pccSTM pcc) + return pcc { pccSTM = new_pcc_stm } + where f_name = "mutatePCC" + +--------------------------------------------------------------------------------------------------------------------------- +-- * PriorityChansConverger I/O +--------------------------------------------------------------------------------------------------------------------------- + +-- | Wrapper around @'writeInPCC_STM'@. Used locks: if blocking, +-- then @[]@ else @['PCC_Lock']@. +-- +-- If blocking is enabled, may block, when there is no free space +-- in @'PriorityChan'@. +writeInPCC :: Ord k => PermitToBlock -> k -> e -> PriorityChansConverger k e -> IO (Maybe FailureReasonWPCC) +writeInPCC block_dowe k e pcc = maybe_lock $ maybeObserveExceptions pcc f_name $ atomically $ writeInPCC_STM block_dowe k e (pccSTM pcc) + where maybe_lock = case block_dowe of { True -> id; False -> withLockedDo f_name [PCC_Lock] pcc } + f_name = "writeInPCC" + +-- | Wrapper around @'interruptableWriteInPCC_STM'@. Used locks: @[]@ +-- +-- Blocking write. +-- +-- Do not put the control of the @'STM' 'InterruptShouldWe'@ under @'PCC_Lock'@. +-- Keep there @False@, if you don't want to get @Left ()@ in return. +-- Put there @True@, whenever you want to stop waiting. +stmInterruptableWriteInPCC :: Ord k => STM InterruptShouldWe -> k -> e -> PriorityChansConverger k e -> IO (Either () (Maybe FailureReasonWPCC)) +stmInterruptableWriteInPCC stm_interrupter k e pcc = maybeObserveExceptions pcc f_name $ atomically $ interruptableWriteInPCC_STM stm_interrupter k e (pccSTM pcc) + where f_name = "stmInterruptableWriteInPCC" + +-- | Wrapper around @'stmInterruptableWriteInPCC'@. Used locks: @['PCC_IOInterruptableWrite']@ +-- +-- Blocking read. +-- +-- Creates a @'TVar' Bool@ to control the interruptability of write operation. +-- Spawns additional thread, which cyclicly reads given @Chan@ and checks, +-- if it signals terminating condition - if so, then using "interrupter" @TVar@ +-- stops trying to write in PCC. Temporary thread gets terminated at the end. +interruptableWriteInPCC :: Ord k => (Chan a, a -> InterruptShouldWe) -> k -> e -> PriorityChansConverger k e -> IO (Either () (Maybe FailureReasonWPCC)) +interruptableWriteInPCC (interrupter_chan, interrupter_cond) k e pcc = withLockedDo f_name [PCC_IOInterruptableWrite] pcc $ maybeObserveExceptions pcc f_name $ do + interrupter_tv <- atomically $ newTVar False + tmp_thread_interrupter <- newIORef False + _ <- forkIO $ let _cycle = do + stop <- readIORef tmp_thread_interrupter + case stop of + True -> return () + False -> do + _interrupt <- interrupter_cond `liftM` readChan interrupter_chan + case _interrupt of + False -> _cycle + True -> atomically $ writeTVar interrupter_tv True + in _cycle + r <- stmInterruptableWriteInPCC (readTVar interrupter_tv) k e pcc + writeIORef tmp_thread_interrupter True + return r + where f_name = "interruptableWriteInPCC" + +-- | Wrapper around @'readFromPCC_STM'@. Used locks: if blocking, then @[]@ else @['PCC_Lock']@. +readFromPCC :: Ord k => PermitToBlock -> PriorityChansConverger k e -> IO (Maybe (k, e)) +readFromPCC block_dowe pcc = maybe_lock $ maybeObserveExceptions pcc f_name $ atomically $ readFromPCC_STM block_dowe (pccSTM pcc) + where maybe_lock = case block_dowe of { True -> id; False -> withLockedDo f_name [PCC_Lock] pcc } + f_name = "readFromPCC" + +-- | Wrapper around @'interruptableReadFromPCC_STM'@. Used locks: @[]@ +-- +-- Blocking read. +stmInterruptableReadFromPCC :: Ord k => STM InterruptShouldWe -> PriorityChansConverger k e -> IO (Either () (k, e)) +stmInterruptableReadFromPCC stm_interrupter pcc = maybeObserveExceptions pcc f_name $ atomically $ interruptableReadFromPCC_STM stm_interrupter (pccSTM pcc) + where f_name = "stmInterruptableReadFromPCC" + +-- | Wrapper around @'stmInterruptableReadFromPCC'@. Used locks: @['PCC_IOInterruptableRead']@ +-- +-- Blocking read. +-- +-- Creates a @'TVar' Bool@ to control the interruptability of read operation. +-- Spawns additional thread, which cyclicly reads given @Chan@ and checks, +-- if it signals terminating condition - if so, then using "interrupter" @TVar@ +-- stops trying to write in PCC. Temporary thread gets terminated at the end. +interruptableReadFromPCC :: Ord k => (Chan a, a -> InterruptShouldWe) -> PriorityChansConverger k e -> IO (Either () (k, e)) +interruptableReadFromPCC (interrupter_chan, interrupter_cond) pcc = withLockedDo f_name [PCC_IOInterruptableRead] pcc $ maybeObserveExceptions pcc f_name $ do + interrupter_tv <- atomically $ newTVar False + tmp_thread_interrupter <- newIORef False + _ <- forkIO $ let _cycle = do + stop <- readIORef tmp_thread_interrupter + case stop of + True -> return () + False -> do + _interrupt <- interrupter_cond `liftM` readChan interrupter_chan + case _interrupt of + False -> _cycle + True -> atomically $ writeTVar interrupter_tv True + in _cycle + r <- stmInterruptableReadFromPCC (readTVar interrupter_tv) pcc + writeIORef tmp_thread_interrupter True + return r + where f_name = "interruptableReadFromPCC" + +-- | Wrapper around @'flushPCC2List_STM'@. Used locks: @['PCC_Lock']@ +-- +-- Lazy (doublecheck that). +flushPCC2List :: Ord k => PriorityChansConverger k e -> IO [(k, e)] +flushPCC2List pcc = withLockedDo f_name [PCC_Lock] pcc $ maybeObserveExceptions pcc f_name $ atomically $ flushPCC2List_STM (pccSTM pcc) + where f_name = "flushPCC2List" + +-- | Wrapper around @flushPCC2List_STM'@ +-- (see "Control.Concurrent.PriorityChansConverger.PriorityChansConverger"). +-- Used locks: @['PCC_Lock']@ +-- +-- Strict. Should be a bit faster than @'flushPCC2List'@. +flushPCC2List' :: Ord k => PriorityChansConverger k e -> IO [(k, e)] +flushPCC2List' pcc = withLockedDo f_name [PCC_Lock] pcc $ maybeObserveExceptions pcc f_name $ atomically $ flushPCC2List_STM' (pccSTM pcc) + where f_name = "flushPCC2List\'" + +-- | Wrapper around @fastFlushPCC2List_STM'@ +-- (see "Control.Concurrent.PriorityChansConverger.PriorityChansConverger"). +-- Used locks: @['PCC_Lock']@ +-- +-- Strict. This is a fast flush, it doesn't deal with priorities. +fastFlushPCC2List' :: Ord k => PriorityChansConverger k e -> IO [(k, e)] +fastFlushPCC2List' pcc = withLockedDo f_name [PCC_Lock] pcc $ maybeObserveExceptions pcc f_name $ atomically $ fastFlushPCC2List_STM' (pccSTM pcc) + where f_name = "fastFlushPCC2List\'" + +-- | Wrapper around @'isEmptyPCC_STM'@. Used locks: none +isEmptyPCC :: Ord k => PriorityChansConverger k e -> IO Bool +isEmptyPCC pcc = maybeObserveExceptions pcc f_name $ atomically $ isEmptyPCC_STM (pccSTM pcc) + where f_name = "isEmptyPCC" + +-- | Wrapper around @'freeSpaceInPCCInput_STM'@. Used locks: none +freeSpaceInPCCInput :: Ord k => PriorityChansConverger k e -> k -> IO (Maybe ChanContentAmountMeasure) +freeSpaceInPCCInput pcc k = maybeObserveExceptions pcc f_name $ atomically $ freeSpaceInPCCInput_STM (pccSTM pcc) k + where f_name = "freeSpaceInPCCInput" + +-- | Wrapper around @filterOutPCCElements_STM'@ +-- (see "Control.Concurrent.PriorityChansConverger.PriorityChansConverger"). +-- Used locks: @['PCC_Lock']@ +-- +-- Strict. +-- +-- Take everything, applu filter, return what's left in the PCC, +-- return what's taken out. +filterOutPCCElements' :: Ord k => PriorityChansConverger k e -> (e -> TakeElementOutShouldWe) -> IO [(k, e)] +filterOutPCCElements' pcc p = withLockedDo f_name [PCC_Lock] pcc $ maybeObserveExceptions pcc f_name $ atomically $ filterOutPCCElements_STM' (pccSTM pcc) p + where f_name = "filterOutPCCElements\'" + +--------------------------------------------------------------------------------------------------------------------------- +-- * PriorityChansConverger representation +--------------------------------------------------------------------------------------------------------------------------- + +-- | Wrapper around @'showPCC'@. Used locks: none +showPCC :: (Ord k, Show k, Show e) => PriorityChansConverger k e -> IO String +showPCC pcc = maybeObserveExceptions pcc f_name $ atomically $ showPCC_STM (pccSTM pcc) + where f_name = "showPCC" diff --git a/Control/Concurrent/PriorityChansConverger/PriorityChansConvergerSTM.hs b/Control/Concurrent/PriorityChansConverger/PriorityChansConvergerSTM.hs new file mode 100644 --- /dev/null +++ b/Control/Concurrent/PriorityChansConverger/PriorityChansConvergerSTM.hs @@ -0,0 +1,428 @@ +{- +Copyright (C) 2009-2010 Andrejs Sisojevs + +All rights reserved. + +For license and copyright information, see the file COPYRIGHT +-} + +-------------------------------------------------------------------------- +-------------------------------------------------------------------------- + +{-# LANGUAGE MultiParamTypeClasses #-} + +-- | Abbreviation for the @PriorityChansConverger@ is PCC. +-- +-- Based on @STM.TChan@s, extended with capacity control +-- (see @"Control.Concurrent.STM.TChan.TChanB"@). +-- +-- When user reads from the PCC, the choice is made - from which channel +-- to read. System selects a nonempty channel, whose /(CurrentPriority, +-- StartPriority)/ tuple is max. The side effect of the channel selection +-- is it's /CurrentPriority/ decrease by one, if it's value becomes +-- less than one, then the /CurrentPriority/ is set to /StartPriority/. +module Control.Concurrent.PriorityChansConverger.PriorityChansConvergerSTM where + +import Control.Concurrent.ConcurrentUISupport +import Control.Concurrent.PriorityChansConverger.Commons +import Control.Concurrent.STM +import Control.Concurrent.STM.TChan.TChanB +import Control.Monad +import Data.Foldable +import qualified Data.Map as M +import Data.Map (Map) +import Data.Maybe +import Data.MyHelpers + +---------------------------------------- +-- * PriorityChansConverger ADT and it's administration + +-- | Decrease @'CurrentPriority'@ by one, if it's value becomes +-- less than one, then the @'CurrentPriority'@ is set to @'StartPriority'@. +decPriority :: (CurrentPriority, StartPriority) -> (CurrentPriority, StartPriority) +decPriority (cur, start) = + ( case cur <= 1 of + True -> start + False -> cur - 1 + , start + ) + +-- Core ADT of this package. It is thought to be wrapped in a IO ADT, +-- though, in order to optimize performance +-- (which is done in the +-- @Control.Concurrent.PriorityChansConverger.PriorityChansConverger@ +-- module). +data Ord k => PriorityChansConvergerSTM k e = PriorityChansConvergerSTM { + -- | All channels. Convenient to use for input. + -- Map keys == @'pcID'@ of elements. + pccStruct :: Map k (PriorityChan k e) + -- | Empty channels. Map keys == @'pcPriorityTV'@ of elements. + , pccEmpties :: TVar (Map (CurrentPriority, StartPriority) [PriorityChan k e]) + -- | Nonempty channels. Map keys == @'pcPriorityTV'@ of elements. + -- Convenient for taking output. + , pccNonEmpties :: TVar (Map (CurrentPriority, StartPriority) [PriorityChan k e]) + -- | For testing and debugging purposes. + , pccConcurrentUISupport :: Maybe ConcurrentUISupport +} + +-- | PriorityChan = ID + @'TChanB'@ + @'TVar' ('CurrentPriority', 'StartPriority')@ +data PriorityChan k e = PriorityChan { + pcID :: k + , pcChanB :: TChanB e + , pcPriorityTV :: TVar (CurrentPriority, StartPriority) +} + +-- | @'pcLoad' = 'chanLoad' . 'pcChanB'@ +pcLoad :: PriorityChan k e -> STM ChanLoad +pcLoad = chanLoad . pcChanB + +-- | @'pcCapacity' = 'chanCapacity' . 'pcChanB'@ +pcCapacity :: PriorityChan k e -> STM ChanCapacity +pcCapacity = chanCapacity . pcChanB + +-- | @'pcIsEmpty' = 'isEmptyTChanB' . 'pcChanB'@ +pcIsEmpty :: PriorityChan k e -> STM Bool +pcIsEmpty = isEmptyTChanB . pcChanB + +-- | @'pcPriority' = 'readTVar' . 'pcPriorityTV'@ +pcPriority :: PriorityChan k e -> STM (CurrentPriority, StartPriority) +pcPriority = readTVar . pcPriorityTV + +-- | @'pcStartPriority' pc = snd `liftM` 'pcPriority' pc@ +pcStartPriority :: PriorityChan k e -> STM StartPriority +pcStartPriority pc = snd `liftM` pcPriority pc + +-- | Constructor. +newPriorityChan :: k -> StartPriority -> ChanCapacity -> STM (PriorityChan k e) +newPriorityChan _id _pcStartPrior _pcCapac = do + _pcChanB <- newTChanB _pcCapac + _pcPriorityTV <- newTVar (_pcStartPrior, _pcStartPrior) + return PriorityChan { + pcID = _id + , pcChanB = _pcChanB + , pcPriorityTV = _pcPriorityTV + } + +-- | PCC constructor with an option to enable utility @'ConcurrentUISupport'@ +-- (which is useful for debugging). +newPriorityChansConvergerSTM_wCUIS :: Ord k => Maybe ConcurrentUISupport -> Map k (StartPriority, ChanCapacity) -> STM (PriorityChansConvergerSTM k e) +newPriorityChansConvergerSTM_wCUIS mb_cuis chan_priors_map = do + (_pccStruct, _pccEmpties) <- + foldrM (\ (k, (p, c)) (struct, state) -> do + pc <- newPriorityChan k p c + let new_struct = M.insert k pc struct + let new_state = M.unionWith (++) state (M.singleton (p, p) [pc]) + return (new_struct, new_state) + ) + (M.empty, M.empty) + (M.toList chan_priors_map) + _pccEmptiesTV <- newTVar _pccEmpties + _pccNonemptiesTV <- newTVar M.empty + return PriorityChansConvergerSTM { + pccStruct = _pccStruct + , pccEmpties = _pccEmptiesTV + , pccNonEmpties = _pccNonemptiesTV + , pccConcurrentUISupport = mb_cuis + } + +-- | PCC constructor. @'ConcurrentUISupport'@ is off. +-- +-- @newPriorityChansConvergerSTM = 'newPriorityChansConvergerSTM_wCUIS' Nothing@ +newPriorityChansConvergerSTM :: Ord k => Map k (StartPriority, ChanCapacity) -> STM (PriorityChansConvergerSTM k e) +newPriorityChansConvergerSTM = newPriorityChansConvergerSTM_wCUIS Nothing + +--------------------------------------------------------------------------------------------------------------------------- +-- * PriorityChansConvergerSTM mutation +--------------------------------------------------------------------------------------------------------------------------- + +-- | Checks: +-- +-- * If PCC contains all channels, constructable by given specification. +-- +-- * If PCC contains nothing else, than mentioned in the given specification. +isOfStructPCC_STM :: Ord k => Map k (StartPriority, ChanCapacity) -> PriorityChansConvergerSTM k e -> STM (PCC_ContainsGivenSpecification, NothingElseIsInStruct) +isOfStructPCC_STM mp pcc = do + (sogood_sofar, left_overs) <- + foldrM (\ (k, (p, c)) (sogood_sofar, mp) -> + case takeFromMap k mp of + (Nothing, _) -> return (False, mp) + (Just pc, rest_mp) -> do + cap <- pcCapacity pc + start <- pcStartPriority pc + return (sogood_sofar && cap == c && start == p, rest_mp) + ) + (True, pccStruct pcc) + (M.toList mp) + return (sogood_sofar, M.null left_overs) + +-- | Updates PCC with given specification. New channels may be added +-- or existing alerted. +-- But no operation for removing channels. +mutatePCC_STM :: Ord k => Map k (StartPriority, ChanCapacity) -> PriorityChansConvergerSTM k e -> STM (PriorityChansConvergerSTM k e) +mutatePCC_STM chan_priors_map pcc = do + new_struct_map <- getNewStruct + return pcc { pccStruct = new_struct_map } + where + getNewStruct = foldrM + (\ (k, (new_start, cap)) struct_map -> do + (new_struct_map, pc, mb_old_prior) <- + case M.lookup k struct_map of + Nothing -> do + pc <- newPriorityChan k new_start cap + return $ (M.insert k pc struct_map, pc, Nothing) + Just pc -> do + (_cur, _start) <- readTVar (pcPriorityTV pc) + let new_prior = (_cur, new_start) + writeTVar (pcPriorityTV pc) (_cur, new_start) + changeChanCapacity cap (pcChanB pc) + return (struct_map, pc, Just (_cur, _start)) + emp <- pcIsEmpty pc + case mb_old_prior of + Nothing -> modifyTVar_ + (pccEmpties pcc) + (M.unionWith (++) (M.singleton (new_start, new_start) [pc])) + Just old_key@(cur, old_start) -> modifyTVar_ + (case emp of + True -> pccEmpties pcc + False -> pccNonEmpties pcc + ) + (\ mp0 -> + let mp1 = M.update (\ pc_list -> + let new_pc_list = filter (\ pc -> pcID pc /= k) pc_list + in case null new_pc_list of + True -> Nothing + False -> Just new_pc_list + ) old_key mp0 + mp2 = M.unionWith (++) (M.singleton (cur, new_start) [pc]) mp1 + in mp2 + ) + return new_struct_map + ) + (pccStruct pcc) + (M.toList chan_priors_map) + +--------------------------------------------------------------------------------------------------------------------------- +-- * PriorityChansConvergerSTM I/O +--------------------------------------------------------------------------------------------------------------------------- + +-- | If blocking is on, then blocks, when trying to write to a full channel +-- (where's no free space). +writeInPCC_STM :: Ord k => PermitToBlock -> k -> e -> PriorityChansConvergerSTM k e -> STM (Maybe FailureReasonWPCC) +writeInPCC_STM block_dowe k e pcc = + case M.lookup k (pccStruct pcc) of + Nothing -> return $ Just BadKey_FRWPCC + Just pc -> do + emp <- pcIsEmpty pc + wrote <- elasticWriteTChanB block_dowe (pcChanB pc) e + case wrote of + False -> return $ Just ChanFull_FRWPCC + True -> + case emp of + False -> return Nothing + True -> do + prior_tuple <- pcPriority pc + modifyTVar_ + (pccEmpties pcc) + (M.update + (\ pc_list -> + case filter (\ pc -> pcID pc /= k) pc_list of + [] -> Nothing + new_pc_list -> Just new_pc_list + ) + prior_tuple + ) + modifyTVar_ + (pccNonEmpties pcc) + (M.unionWith (++) (M.singleton prior_tuple [pc])) + return Nothing + +-- | Wrapper around @('writeInPCC_STM' True)@ - blocking write. +-- Uses 'interruptableSTM'. If returns Left, then transaction ended with +-- interrupting condition. +interruptableWriteInPCC_STM :: Ord k => STM InterruptShouldWe -> k -> e -> PriorityChansConvergerSTM k e -> STM (Either () (Maybe FailureReasonWPCC)) +interruptableWriteInPCC_STM stm_interrupter k e pcc = do + mb_r <- interruptableSTM stm_interrupter (writeInPCC_STM True k e pcc) + return $ case mb_r of + Nothing -> Left () + Just smtng -> Right smtng + +-- | If @'PermitToBlock'@ is @True@, then never returns @Nothing@. +readFromPCC_STM :: Ord k => PermitToBlock -> PriorityChansConvergerSTM k e -> STM (Maybe (k, e)) +readFromPCC_STM block_dowe pcc = do + mp <- readTVar $ pccNonEmpties pcc + case M.maxViewWithKey mp of + Nothing -> + case block_dowe of + True -> retry + False -> return Nothing + Just ((prior, pc : rest_pc_list), rest_mp) -> do -- this assumes that on place of "pc : _" can't be [] + mb_e <- fromJust `liftM` tryReadTChanB (pcChanB pc) -- this assumes that if given PriorityChan is in pccNonEmpties map, then it may NOT be empty... orelse, we get an exception here + emp <- pcIsEmpty pc + let new_prior = decPriority prior + case emp of + True -> modifyTVar_ + (pccEmpties pcc) + (M.unionWith (++) (M.singleton new_prior [pc])) + False -> return () + let new_mp = + case emp of + True -> + case rest_pc_list of + [] -> rest_mp + _ -> M.insert prior rest_pc_list rest_mp + False -> + let map_1 = + case rest_pc_list of + [] -> rest_mp + _ -> M.insert prior rest_pc_list rest_mp + in M.unionWith (++) (M.singleton new_prior [pc]) map_1 + modifyTVar_ + (pccNonEmpties pcc) + (const new_mp) + modifyTVar_ + (pcPriorityTV pc) + (const new_prior) + return $ Just (pcID pc, mb_e) + Just ((_, []),_) -> error "An error occurred in 'readFromPCC_STM': empty list in the entry of pccNonEmpties!" + where f_name = "readFromPCC_STM" + +-- | Wrapper around @('readFromPCC_STM' True)@ - blocking read. +-- Uses 'interruptableSTM'. If returns Left, then transaction ended with +-- interrupting condition. +interruptableReadFromPCC_STM :: Ord k => STM InterruptShouldWe -> PriorityChansConvergerSTM k e -> STM (Either () (k, e)) +interruptableReadFromPCC_STM stm_interrupter pcc = do + mb_r <- interruptableSTM stm_interrupter (readFromPCC_STM True pcc) + return $ case mb_r of + Nothing -> Left () + Just (Just smtng) -> Right smtng + Just Nothing -> error "An error occurred in 'interruptableReadFromPCC_STM': blocking read returned Nothing!" + +-- | Composition of @'readFromPCC_STM'@s. Lazy (doublecheck that). +flushPCC2List_STM :: Ord k => PriorityChansConvergerSTM k e -> STM [(k, e)] +flushPCC2List_STM pcc = reverse `liftM` whileJustM_1 (:) [] (readFromPCC_STM False pcc) + +-- | Strict. Should be a bit faster than @'flushPCC2List_STM'@. +flushPCC2List_STM' :: Ord k => PriorityChansConvergerSTM k e -> STM [(k, e)] +flushPCC2List_STM' pcc = do + ne_pc_mp <- readTVar $ pccNonEmpties pcc + (e_list, empties) <- readFromNonempties ne_pc_mp + modifyTVar_ (pccEmpties pcc) (M.unionWith (++) empties) + return e_list + where + readFromNonempties :: Map (CurrentPriority, StartPriority) [PriorityChan k e] -> STM ([(k, e)], Map (CurrentPriority, StartPriority) [PriorityChan k e]) + readFromNonempties mp = + case M.maxViewWithKey mp of + Nothing -> return ([], M.empty) + Just ((prior, pc_list), rest_mp) -> do + let new_prior = decPriority prior + (e_list, nonempties, empties) <- foldrM + (\ pc (elems_accum, nonempties_accum, empties_accum) -> do + mb_e <- tryReadTChanB (pcChanB pc) + return $ case mb_e of + Nothing -> ( elems_accum, nonempties_accum, pc : empties_accum) + Just e -> ((pcID pc, e) : elems_accum, pc : nonempties_accum, empties_accum) + ) + ([], [], []) + pc_list + let continue_with_map = M.unionWith (++) rest_mp (M.singleton new_prior nonempties) + let stay_with_map = M.singleton prior empties + (lower_prior_e_list, rest_of_empties) <- readFromNonempties continue_with_map + return (e_list ++ lower_prior_e_list, M.unionWith (++) stay_with_map rest_of_empties) + +-- | Strict. This is a fast flush, it doesn't deal with priorities. +fastFlushPCC2List_STM' :: Ord k => PriorityChansConvergerSTM k e -> STM [(k, e)] +fastFlushPCC2List_STM' pcc = do + ne_pc_mp <- readTVar $ pccNonEmpties pcc + modifyTVar_ (pccEmpties pcc) (M.unionWith (++) ne_pc_mp) + let pc_lists_list = M.elems ne_pc_mp + elems_list <- foldlM (foldlM foldlM_r) [] pc_lists_list + return elems_list + where + foldlM_r :: [(k, e)] -> PriorityChan k e -> STM [(k, e)] + foldlM_r accum pc = do + let k = pcID pc + el <- getTChanBContents (pcChanB pc) + return (accum ++ map (\ e -> (k, e)) el) + +isEmptyPCC_STM :: Ord k => PriorityChansConvergerSTM k e -> STM Bool +isEmptyPCC_STM pcc = M.null `liftM` (readTVar $ pccNonEmpties pcc) + +-- | Free space in referenced channel = Capacity - Load. +freeSpaceInPCCInput_STM :: Ord k => PriorityChansConvergerSTM k e -> k -> STM (Maybe ChanContentAmountMeasure) +freeSpaceInPCCInput_STM pcc k = do + case M.lookup k $ pccStruct pcc of + Nothing -> return Nothing + Just pc -> do + cap <- pcCapacity pc + lo <- pcLoad pc + return $ Just (cap - lo) + +-- | Strict. +-- +-- Take everything, applu filter, return what's left in the PCC, +-- return what's taken out. +filterOutPCCElements_STM' :: Ord k => PriorityChansConvergerSTM k e -> (e -> TakeElementOutShouldWe) -> STM [(k, e)] +filterOutPCCElements_STM' pcc p = foldlM _filterOutPCElements [] (M.elems $ pccStruct pcc) + where + -- _filterOutPCElements :: [(k, e)] -> PriorityChan k e -> STM [(k, e)] + _filterOutPCElements takeout_accum pc = do + takeout_add <- filterOutTChanBElements p (pcChanB pc) + return $ join [zip (repeat $ pcID pc) takeout_add, takeout_accum] + + +--------------------------------------------------------------------------------------------------------------------------- +-- * PriorityChansConvergerSTM representation +--------------------------------------------------------------------------------------------------------------------------- + +-- | Used wrapped in @(unsafePerformIO . atomically)@ in the definition +-- of show instance. +showPCC_STM :: (Ord k, Show k, Show e) => PriorityChansConvergerSTM k e -> STM String +showPCC_STM pcc = do + let f = (\ pc_map_by_prior -> map (\ (p, pc_list) -> (p, map pcID pc_list)) (M.toList pc_map_by_prior) ) + emps <- f `liftM` (readTVar $ pccEmpties pcc) + nemps <- f `liftM` (readTVar $ pccNonEmpties pcc) + let s = "PriorityChansConvergerSTM {" + ++ "\n | Empties = " ++ (show $ emps) + ++ "\n | NonEmpties = " ++ (show $ nemps) + ++ "\n | Chans count = " ++ (show $ M.size $ pccStruct pcc) ++ " chans" + ++ "\n |+ Channels : " + s_w_c <- foldrM (\ pc s_accum -> showPC_STM pc >>= \ s_add -> return (s_accum ++ s_add) ) s (M.elems $ pccStruct pcc) + return (s_w_c ++ "\n}") + +-- | Used wrapped in @(unsafePerformIO . atomically)@ in the definition +-- of show instance. +showPC_STM :: (Show e, Show k) => PriorityChan k e -> STM String +showPC_STM pc = do + (cur, start) <- pcPriority pc + cap <- pcCapacity pc + load <- pcLoad pc + let s = "\n ||+ PriorityChan {" + ++ "\n ||| ID = " ++ (show $ pcID pc) + ++ "\n ||| CurrentPriority = " ++ (show cur) + ++ "\n ||| StartPriority = " ++ (show start) + ++ "\n ||| Capacity = " ++ (show cap) + ++ "\n ||| Load = " ++ (show load) + ++ "\n |||+ Content : " + elems <- viewChanBContent $ pcChanB pc + let ch_content_str = (Prelude.concat $ mapWithIndex (\ i e -> "\n |||| -- " ++ show i ++ ") " ++ show e) elems) + ++ "\n ||- }" + return (s ++ ch_content_str) + +-- instance (Show k, Show e) => Show (PriorityChan k e) where +-- show pc = unsafePerformIO $ printExceptionIfAny "show:PriorityChan" $ atomically $ showPC_STM pc +-- instance (Ord k, Show k, Show e) => Show (PriorityChansConvergerSTM k e) where +-- show pcc = unsafePerformIO $ printExceptionIfAny "show:PriorityChansConvergerSTM" $ atomically $ showPCC_STM pcc + +-------------------------------- + +-- * Helpers + +-- | For @interruptableSTM interrupter subj@, when @interrupter@ returns @True@, +-- the transaction returns @Nothing@. Else the result of @subj@ is waited +-- and returned wrapped into @Just@. +interruptableSTM :: STM InterruptShouldWe -> STM a -> STM (Maybe a) +interruptableSTM interrupter subj = do + ir <- interrupter + case ir of + True -> return Nothing + False -> Just `liftM` subj diff --git a/Control/Concurrent/STM/TChan/ExtsCommons.hs b/Control/Concurrent/STM/TChan/ExtsCommons.hs new file mode 100644 --- /dev/null +++ b/Control/Concurrent/STM/TChan/ExtsCommons.hs @@ -0,0 +1,22 @@ +{- +Copyright (C) 2009-2010 Andrejs Sisojevs + +All rights reserved. + +For license and copyright information, see the file COPYRIGHT +-} + +-------------------------------------------------------------------------- +-------------------------------------------------------------------------- + +module Control.Concurrent.STM.TChan.ExtsCommons where + +-- | Bigger than @Int@ might not work, because some list functions are used, +-- that consume @Int@ +type ChanContentAmountMeasure = Int +type ChanLoad = ChanContentAmountMeasure +type ChanCapacity = ChanContentAmountMeasure +type PermitToBlock = Bool +type WriteSucceeded = Bool +type TakeElementOutShouldWe = Bool + diff --git a/Control/Concurrent/STM/TChan/TChanB.hs b/Control/Concurrent/STM/TChan/TChanB.hs new file mode 100644 --- /dev/null +++ b/Control/Concurrent/STM/TChan/TChanB.hs @@ -0,0 +1,18 @@ +{- +Copyright (C) 2009-2010 Andrejs Sisojevs + +All rights reserved. + +For license and copyright information, see the file COPYRIGHT +-} + +-------------------------------------------------------------------------- +-------------------------------------------------------------------------- + +module Control.Concurrent.STM.TChan.TChanB ( + module Control.Concurrent.STM.TChan.TChanB_ + , module Control.Concurrent.STM.TChan.ExtsCommons + ) where + +import Control.Concurrent.STM.TChan.TChanB_ +import Control.Concurrent.STM.TChan.ExtsCommons diff --git a/Control/Concurrent/STM/TChan/TChanB_.hs b/Control/Concurrent/STM/TChan/TChanB_.hs new file mode 100644 --- /dev/null +++ b/Control/Concurrent/STM/TChan/TChanB_.hs @@ -0,0 +1,127 @@ +{- +Copyright (C) 2009-2010 Andrejs Sisojevs + +All rights reserved. + +For license and copyright information, see the file COPYRIGHT +-} + +-------------------------------------------------------------------------- +-------------------------------------------------------------------------- + +-- | @'TChanL' = 'TChan' + 'TVar' 'ChanLoad'@ +-- +-- @'TChanB' = 'TChanL' + 'TVar' 'ChanCapacity'@ +module Control.Concurrent.STM.TChan.TChanB_ where + +import Control.Concurrent +import Control.Concurrent.STM +import Control.Concurrent.STM.TChan.TChanL hiding (chanLoad) +import qualified Control.Concurrent.STM.TChan.TChanL as TCHL (chanLoad) +import Data.List +import Data.MyHelpers + +-------------------------------------------------------------- + +data TChanB a = + TChanB { + tcbChanL :: TChanL a + , tcbCapacity :: TVar ChanCapacity + } + +chanCapacity :: TChanB a -> STM ChanCapacity +chanCapacity tchb = readTVar $ tcbCapacity tchb + +changeChanCapacity :: ChanCapacity -> TChanB a -> STM () +changeChanCapacity cap tchb = writeTVar (tcbCapacity tchb) cap + +chanLoad :: TChanB a -> STM ChanCapacity +chanLoad tchb = TCHL.chanLoad $ tcbChanL tchb + +newTChanB :: ChanCapacity -> STM (TChanB a) +newTChanB cap = do + tch <- newTChanL + tv <- newTVar cap + return TChanB { + tcbChanL = tch + , tcbCapacity = tv + } + +newTChanB_IO :: ChanCapacity -> IO (TChanB a) +newTChanB_IO cap = atomically $ newTChanB cap + +-- | The \"elastic-\" prefix stands for the opportunity to regulate +-- with option 'PermitToBlock' +elasticWriteTChanB :: PermitToBlock -> TChanB a -> a -> STM WriteSucceeded +elasticWriteTChanB block_dowe tchb a = do + load <- chanLoad tchb + cap <- chanCapacity tchb + case load < cap of + True -> return True << writeTChanL (tcbChanL tchb) a + False -> case block_dowe of + True -> retry + False -> return False + +-- | Wrapper around 'elasticWriteTChanB': +-- @writeTChanB tchb a = return () << elasticWriteTChanB True tchb a@ +writeTChanB :: TChanB a -> a -> STM () +writeTChanB tchb a = return () << elasticWriteTChanB True tchb a + +readTChanB :: TChanB a -> STM a +readTChanB tchb = readTChanL (tcbChanL tchb) + +isEmptyTChanB :: TChanB a -> STM Bool +isEmptyTChanB tchb = isEmptyTChanL (tcbChanL tchb) + +-- | @freeSpace = 'chanCapacity' - 'chanLoad'@ +freeSpaceInTChanB :: TChanB a -> STM ChanContentAmountMeasure +freeSpaceInTChanB tchb = do + load <- chanLoad tchb + cap <- chanCapacity tchb + return (cap - load) + +tryReadTChanB :: TChanB a -> STM (Maybe a) +tryReadTChanB tchb = tryReadTChanL $ tcbChanL tchb + +-- | Wrapper around 'elasticWriteTChanB': +-- @tryWriteTChanB = elasticWriteTChanB False@ +tryWriteTChanB :: TChanB a -> a -> STM Bool +tryWriteTChanB = elasticWriteTChanB False + +-- | Take everything from channel, and leave it empty. +getTChanBContents :: TChanB a -> STM [a] +getTChanBContents = getTChanLContents . tcbChanL + +-- | This procedure is highly uneffective and is dangerous, +-- when @'PermitToBlock'@ is on. +-- Since it won't complete unless there is enough free space in the channnel. +-- Even the part of input that fits won't commit to get into the channel. +-- If capacity isn't enough, it won't ever commit, +-- unless the capacity is made bigger. +elacticWriteList2TChanB :: PermitToBlock -> TChanB a -> [a] -> STM [a] +elacticWriteList2TChanB block_dowe tchb l = do + available_space <- freeSpaceInTChanB tchb + let (to_load, rest) = splitAt available_space l + case block_dowe of + False -> do + writeList2TChanL (tcbChanL tchb) to_load + return rest + True -> + case rest of + [] -> do + writeList2TChanL (tcbChanL tchb) to_load + return [] + _ -> retry + +-- | Just filter out elements. Is abstracted over implementation, +-- so doesn't suffer from possible inconsistency between capacity +-- and load (when capacity < load). +filterOutTChanBElements :: (a -> TakeElementOutShouldWe) -> TChanB a -> STM [a] +filterOutTChanBElements p tchb = do + l <- getTChanBContents tchb + let (takeout, stay) = partition p l + didnt_fit_back <- writeList2TChanL (tcbChanL tchb) stay + return takeout + +viewChanBContent :: TChanB a -> STM [a] +viewChanBContent tchb = viewChanLContent (tcbChanL tchb) diff --git a/Control/Concurrent/STM/TChan/TChanL.hs b/Control/Concurrent/STM/TChan/TChanL.hs new file mode 100644 --- /dev/null +++ b/Control/Concurrent/STM/TChan/TChanL.hs @@ -0,0 +1,19 @@ +{- +Copyright (C) 2009-2010 Andrejs Sisojevs + +All rights reserved. + +For license and copyright information, see the file COPYRIGHT +-} + +-------------------------------------------------------------------------- +-------------------------------------------------------------------------- + + +module Control.Concurrent.STM.TChan.TChanL ( + module Control.Concurrent.STM.TChan.TChanL_ + , module Control.Concurrent.STM.TChan.ExtsCommons + ) where + +import Control.Concurrent.STM.TChan.TChanL_ +import Control.Concurrent.STM.TChan.ExtsCommons diff --git a/Control/Concurrent/STM/TChan/TChanL_.hs b/Control/Concurrent/STM/TChan/TChanL_.hs new file mode 100644 --- /dev/null +++ b/Control/Concurrent/STM/TChan/TChanL_.hs @@ -0,0 +1,88 @@ +{- +Copyright (C) 2009-2010 Andrejs Sisojevs + +All rights reserved. + +For license and copyright information, see the file COPYRIGHT +-} + +-------------------------------------------------------------------------- +-------------------------------------------------------------------------- + + +-- | @'TChanL' = 'TChan' + 'TVar' 'ChanLoad'@ +module Control.Concurrent.STM.TChan.TChanL_ where + +import Control.Concurrent +import Control.Concurrent.STM +import Control.Concurrent.STM.TChan.ExtsCommons +import Control.Monad +import Data.Foldable +import Data.List +import Data.MyHelpers + +-------------------------------------------------------------- + +data TChanL a = + TChanL { + tclChan :: TChan a + , tclLoad :: TVar ChanLoad + } + +chanLoad :: TChanL a -> STM ChanLoad +chanLoad tchl = readTVar $ tclLoad tchl + +newTChanL :: STM (TChanL a) +newTChanL = do + tch <- newTChan + tv <- newTVar 0 + return TChanL { + tclChan = tch + , tclLoad = tv + } + +newTChanL_IO :: IO (TChanL a) +newTChanL_IO = atomically newTChanL + +writeTChanL :: TChanL a -> a -> STM () +writeTChanL tchl a = do + modifyTVar_ (tclLoad tchl) (+ 1) + writeTChan (tclChan tchl) a + +readTChanL :: TChanL a -> STM a +readTChanL tchl = do + modifyTVar_ (tclLoad tchl) (\ v -> v - 1) + readTChan (tclChan tchl) + +isEmptyTChanL :: TChanL a -> STM Bool +isEmptyTChanL tchl = isEmptyTChan (tclChan tchl) + +tryReadTChanL :: TChanL a -> STM (Maybe a) +tryReadTChanL tchl = do + emp <- isEmptyTChanL tchl + case emp of + True -> return Nothing + False -> Just `liftM` readTChanL tchl + +getTChanLContents :: TChanL a -> STM [a] +getTChanLContents tchl = do + writeTVar (tclLoad tchl) 0 + reverse `liftM` whileJustM_1 (:) [] (tryReadTChan (tclChan tchl)) + +writeList2TChanL :: TChanL a -> [a] -> STM () +writeList2TChanL tchl l = do + load_add <- foldlM (\ load_accum e -> writeTChan (tclChan tchl) e >> return (load_accum + 1)) 0 l + modifyTVar_ (tclLoad tchl) (\ v -> v + load_add) + +filterOutTChanLElements :: (a -> TakeElementOutShouldWe) -> TChanL a -> STM [a] +filterOutTChanLElements p tchl = do + l <- getTChanLContents tchl + let (takeout, stay) = partition p l + writeList2TChanL tchl stay + return takeout + +viewChanLContent :: TChanL a -> STM [a] +viewChanLContent tchl = do + l <- getTChanLContents tchl + writeList2TChanL tchl l -- I would gladly use some kind of abortChangesSTM here instead of this + return l diff --git a/Data/MyHelpers.hs b/Data/MyHelpers.hs new file mode 100644 --- /dev/null +++ b/Data/MyHelpers.hs @@ -0,0 +1,18 @@ +{- +Copyright (C) 2009-2010 Andrejs Sisojevs + +All rights reserved. + +For license and copyright information, see the file COPYRIGHT +-} + +-------------------------------------------------------------------------- +-------------------------------------------------------------------------- + +module Data.MyHelpers ( + module Data.MyHelpers_ + , module Debug.Trace + ) where + +import Data.MyHelpers_ +import Debug.Trace diff --git a/Data/MyHelpers_.hs b/Data/MyHelpers_.hs new file mode 100644 --- /dev/null +++ b/Data/MyHelpers_.hs @@ -0,0 +1,384 @@ +{- +Copyright (C) 2009-2010 Andrejs Sisojevs + +All rights reserved. + +For license and copyright information, see the file COPYRIGHT +-} + +-------------------------------------------------------------------------- +-------------------------------------------------------------------------- + +{-# LANGUAGE MultiParamTypeClasses, FlexibleInstances #-} + +module Data.MyHelpers_ where + +import Control.Monad +import Control.Concurrent +import Control.Concurrent.STM +import Control.Exception +import Data.Char +import Data.Int +import Data.List +import qualified Data.Map as M +import Data.Map (Map, (!)) +import Data.Maybe +import Data.Typeable +import Data.Word + +import Foreign.Marshal.Alloc +import Foreign.Ptr +import Foreign.Storable + +import System.IO +import System.IO.Unsafe + +-------------------------------------- + +class EQ a b +instance EQ a a + +type ErrorMessage = String + +infixr 1 << +(<<) :: Monad m => m b -> m a -> m b +f << x = x >> f + +reverse2ple :: (a,b) -> (b,a) +reverse2ple (a,b) = (b,a) + +fst3 :: (a,b,c) -> a +snd3 :: (a,b,c) -> b +thrd3 :: (a,b,c) -> c +fst3 (a,_,_) = a +snd3 (_,b,_) = b +thrd3 (_,_,c) = c + +apFor2ple :: (a -> a -> a, b -> b -> b) -> (a,b) -> (a,b) -> (a,b) +apFor2ple (af,bf) (a1,b1) (a2,b2) = (af a1 a2, bf b1 b2) + +liftTuple :: Monad m => (m a, m b) -> m (a, b) +lift2ple :: Monad m => (m a, m b) -> m (a, b) +lift3ple :: Monad m => (m a, m b, m c) -> m (a, b, c) +lift4ple :: Monad m => (m a, m b, m c, m d) -> m (a, b, c, d) +liftTuple (ma, mb) = do a <- ma + b <- mb + return (a,b) +lift2ple = liftTuple +lift3ple (ma, mb, mc) = do a <- ma + b <- mb + c <- mc + return (a,b,c) +lift4ple (ma, mb, mc, md) = do a <- ma + b <- mb + c <- mc + d <- md + return (a,b,c,d) + +ap22ple :: (a, a) -> (a -> b) -> (b, b) +ap22ple (a1,a2) fa = (fa a1, fa a2) +ap22pleM :: Monad m => (a, a) -> (a -> m b) -> m (b, b) +ap22pleM ple f = lift2ple $ ap22ple ple f +ap23ple :: (a, a, a) -> (a -> b) -> (b, b, b) +ap23ple (a1,a2,a3) fa = (fa a1, fa a2, fa a3) +ap23pleM :: Monad m => (a, a, a) -> (a -> m b) -> m (b, b, b) +ap23pleM ple f = lift3ple $ ap23ple ple f +ap24ple :: (a, a, a, a) -> (a -> b) -> (b, b, b, b) +ap24ple (a1,a2,a3,a4) fa = (fa a1, fa a2, fa a3, fa a3) +ap24pleM :: Monad m => (a, a, a, a) -> (a -> m b) -> m (b, b, b, b) +ap24pleM ple f = lift4ple $ ap24ple ple f + +apFrom2ple :: ((a -> b), (a -> c)) -> a -> (b, c) +apFrom2ple (f1,f2) a = (f1 a, f2 a) +apFrom3ple :: ((a -> b), (a -> c), (a -> d)) -> a -> (b, c, d) +apFrom3ple (f1,f2,f3) a = (f1 a, f2 a, f3 a) +apFrom4ple :: ((a -> b), (a -> c), (a -> d), (a -> e)) -> a -> (b, c, d, e) +apFrom4ple (f1,f2,f3,f4) a = (f1 a, f2 a, f3 a, f4 a) + +liftEither6 :: Either er a -> Either er b -> Either er c -> Either er d -> Either er e -> Either er f -> Either er (a,b,c,d,e,f) +liftEither6 er_or_a er_or_b er_or_c er_or_d er_or_e er_or_f = + case er_or_a of + Left er -> Left er + Right a -> + case er_or_b of + Left er -> Left er + Right b -> + case er_or_c of + Left er -> Left er + Right c -> + case er_or_d of + Left er -> Left er + Right d -> + case er_or_e of + Left er -> Left er + Right e -> + case er_or_f of + Left er -> Left er + Right f -> Right (a,b,c,d,e,f) + +liftEither5 :: Either er a -> Either er b -> Either er c -> Either er d -> Either er e -> Either er (a,b,c,d,e) +liftEither5 er_or_a er_or_b er_or_c er_or_d er_or_e = + case er_or_a of + Left er -> Left er + Right a -> + case er_or_b of + Left er -> Left er + Right b -> + case er_or_c of + Left er -> Left er + Right c -> + case er_or_d of + Left er -> Left er + Right d -> + case er_or_e of + Left er -> Left er + Right e -> Right (a,b,c,d,e) + +liftEither2 :: Either er a -> Either er b -> Either er (a,b) +liftEither2 er_or_a er_or_b = + case er_or_a of + Left er -> Left er + Right a -> + case er_or_b of + Left er -> Left er + Right b -> Right (a,b) + +isLeft :: Either a b -> Bool +isLeft (Left _) = True +isLeft _ = False + +isRight :: Either a b -> Bool +isRight = not . isLeft + +fromLeft :: Either a b -> a +fromLeft (Left a) = a +fromLeft _ = error "Error occurred, when applied fromLeft to Right!" + +fromRight :: Either a b -> b +fromRight (Right b) = b +fromRight _ = error "Error occurred, when applied fromLeft to Right!" + +mapWithIndex :: (Int -> a -> b) -> [a] -> [b] +mapWithIndex f l = _mapWithIndex 1 l + where + _mapWithIndex _ [] = [] + _mapWithIndex i (h:t) = f i h : _mapWithIndex i t + +whileJustM_0 :: Monad m => m (Maybe a) -> m [a] +whileJustM_0 mma = mma >>= \ ma -> case ma of + Nothing -> return [] + Just a -> (:) a `liftM` whileJustM_0 mma + +whileJustM_1 :: Monad m => (a -> b -> b) -> b -> m (Maybe a) -> m b +whileJustM_1 accum_f start_accum src = _whileJustM_1 start_accum + where + _whileJustM_1 cur_accum = do + mb_a <- src + case mb_a of + Nothing -> return cur_accum + Just a -> _whileJustM_1 (accum_f a cur_accum) + +takeFromList :: (a -> Bool) -> [a] -> (Maybe a, [a]) +takeFromList p l = _takeFromList l + where + _takeFromList [] = (Nothing, []) + _takeFromList (h:t) = + case p h of + True -> (Just h, t) + False -> let (mb_e, processed_t) = _takeFromList t + in (mb_e, h : processed_t) + +takeFromMap :: Ord k => k -> Map k a -> (Maybe a, Map k a) +takeFromMap k m = + let mb_a = M.lookup k m + in ( mb_a + , case isJust mb_a of + True -> M.delete k m + False -> m + ) + +partition_2 :: Ord k => (a -> Bool) -> (a -> b, a -> c) -> Map k a -> (Map k b, Map k c) +partition_2 p (true_f, false_f) m = + M.foldWithKey + (\ k a (true_map, false_map) -> + case p a of + True -> (M.insert k (true_f a) true_map, false_map) + False -> (true_map, M.insert k (false_f a) false_map) + ) + (M.empty, M.empty) + m + +unionWithKey_2 :: Ord k => (k -> (a, a) -> (a, b)) -> (b -> c -> c) -> c -> Map k a -> Map k a -> (Map k a, c) +unionWithKey_2 unifier side_result_accum_f accum_start m1 m2 = + foldr (\ (k, a1) (m_accum, side_result_accum) -> + case takeFromMap k m_accum of + (Nothing, _) -> (M.insert k a1 m_accum, side_result_accum) + (Just a2, rest_of_accum_map) -> + let (ar, side_result_add) = unifier k (a1, a2) + in (M.insert k ar rest_of_accum_map, side_result_accum_f side_result_add side_result_accum) + ) + (m2, accum_start) + (M.toList m1) + +unionWithKey_3 :: Ord k => (k -> (Maybe a, Maybe b) -> (c, d)) -> (d -> d -> d) -> d -> Map k a -> Map k b -> (Map k c, d) +unionWithKey_3 unifierF sideResultAccumF accum_start m1 m2 = + let (m1_result, accum1, m2_cut) = foldr + (\ (k, a) (m_accum, side_result_accum, m2_cut) -> + let (mb_b, new_m2_cut) = takeFromMap k m2_cut + (c, d) = unifierF k (Just a, mb_b) + new_side_result_accum = sideResultAccumF side_result_accum d + in (M.insert k c m_accum, new_side_result_accum, new_m2_cut) + ) + (M.empty, accum_start, m2) + (M.toList m1) + (m2_result, accum2) = foldr + (\ (k, b) (m_accum, side_result_accum) -> + let (c, d) = unifierF k (Nothing, Just b) + new_side_result_accum = sideResultAccumF side_result_accum d + in (M.insert k c m_accum, new_side_result_accum) + ) + (m1_result, accum1) + (M.toList m2_cut) + in (m2_result, accum2) + +unionWithKey_4 :: Ord k => (k -> (Maybe a, Maybe b) -> (Maybe c, d)) -> (d -> d -> d) -> d -> Map k a -> Map k b -> (Map k c, d) +unionWithKey_4 unifierF sideResultAccumF accum_start m1 m2 = + let (m1_result, accum1, m2_cut) = foldr + (\ (k, a) (m_accum, side_result_accum, m2_cut) -> + let (mb_b, new_m2_cut) = takeFromMap k m2_cut + (mb_c, d) = unifierF k (Just a, mb_b) + new_side_result_accum = sideResultAccumF side_result_accum d + in ( case mb_c of + Just c -> M.insert k c m_accum + Nothing -> m_accum + , new_side_result_accum, new_m2_cut + ) + ) + (M.empty, accum_start, m2) + (M.toList m1) + (m2_result, accum2) = foldr + (\ (k, b) (m_accum, side_result_accum) -> + let (mb_c, d) = unifierF k (Nothing, Just b) + new_side_result_accum = sideResultAccumF side_result_accum d + in ( case mb_c of + Just c -> M.insert k c m_accum + Nothing -> m_accum + , new_side_result_accum + ) + ) + (m1_result, accum1) + (M.toList m2_cut) + in (m2_result, accum2) + +lookupDeleteFromList :: Eq a => a -> [a] -> (Bool, [a]) +lookupDeleteFromList e l = + let mship = elem e l + in ( mship + , case mship of + True -> delete e l + False -> l + ) + +from2DList :: Ord k => [(k, a)] -> M.Map k [a] +from2DList l = foldl (\ accum_map (k, e) -> M.unionWith (++) accum_map (M.singleton k [e])) M.empty l + +str2Numeric :: (Num a, Ord a) => String -> Maybe a +str2Numeric s = _str2Numeric s 0 + where + _str2Numeric [] i = Just i + _str2Numeric (sh:st) i | isDigit sh = let nexti = (i*10) + (fromIntegral $ digitToInt sh) + in case i > nexti of + False -> _str2Numeric st nexti + True -> Nothing + | otherwise = Nothing + +truncLiterary :: String -> Int -> String +truncLiterary str n = case length str > n of + True -> take (n - 3) str ++ "..." + False -> str + +readKeyPress :: IO () +readKeyPress = do + buf <- malloc :: IO (Ptr Word32) + hWaitForInput stdin (negate 1) + hGetBufNonBlocking stdin buf 4 + free buf + +pressAnyKey :: IO () +pressAnyKey = putStrLn "" << readKeyPress << putStr "\n-------Press any key to continue...-------" + +bool2num :: Num a => Bool -> a +bool2num False = 0 +bool2num True = 1 + +whenJust :: Maybe (IO ()) -> IO () +whenJust mb_ioa = whenJustA () mb_ioa + +whenJustA :: a -> Maybe (IO a) -> IO a +whenJustA a mb_ioa = + case mb_ioa of + Nothing -> return a + Just ioa -> ioa + +------------------------------------------ +-- STM + +modifyTVar :: TVar a -> (a -> (a, b)) -> STM b +modifyTVar tv f = do + v <- readTVar tv + let (a, b) = f v + writeTVar tv a + return b + +modifyTVar_ :: TVar a -> (a -> a) -> STM () +modifyTVar_ tv f = do + v <- readTVar tv + writeTVar tv (f v) + +tryReadTChan :: TChan a -> STM (Maybe a) +tryReadTChan tch = do + emp <- isEmptyTChan tch + case emp of + True -> return Nothing + False -> Just `liftM` readTChan tch + +----------------------------- +-- few public helpers + +-- data System.IO.IOMode = ReadMode | WriteMode | AppendMode | ReadWriteMode +dump :: Show a => a -> Int -> IOMode -> IO () +dump a i m = do + h <- openFile ("./dump.out" ++ show i ++ ".hs") m + hPutStr h $ show a + hClose h + +invDump :: Show a => Int -> IOMode -> a -> a +invDump i m a = unsafePerformIO $ do + dump a i m + return a + +watch :: Show a => a -> a +watch a = unsafePerformIO $ do + putStrLn_paged 22 $ show a + return a + +watchCond :: Show a => Bool -> a -> a +watchCond cond = if cond then watch else id + +putStrLn_paged :: Int -> String -> IO () +putStrLn_paged page_size s = f $ lines s + where + f lines_list = + let (to_print, to_next_itera) = splitAt page_size lines_list + in do putStrLn (concat $ intersperse "\n" to_print) + case null to_next_itera of + True -> return () + False -> f to_next_itera << hGetChar stdin << putStrLn "\n-------Press any key to continue...-------" + +traceShowPaged :: Show a => Int -> a -> b -> b +traceShowPaged n a b = unsafePerformIO $ do + putStrLn_paged n $ show a + return b + +traceCond :: Show a => Bool -> Int -> a -> b -> b +traceCond cond n a b = if cond then traceShowPaged n a b else b diff --git a/PriorityChansConverger.cabal b/PriorityChansConverger.cabal new file mode 100644 --- /dev/null +++ b/PriorityChansConverger.cabal @@ -0,0 +1,56 @@ +Name: PriorityChansConverger +Synopsis: + Read single output from an array of inputs - channels with priorities +Description: + Abbreviation for the @PriorityChansConverger@ is PCC. + . + Based on @STM.TChan@, extended with capacity control. + Wrapped into IO version is also available. + . + When user reads from the PCC, the choice is made - from which channel + to read. System selects a nonempty channel, whose (CurrentPriority, + StartPriority) tuple is max. The side effect of the channel selection + is it's /CurrentPriority/ decrease by one, if it's value becomes + less than one, then the /CurrentPriority/ is set to /StartPriority/. + . + For the main API interface: + . + @import Control.Concurrent.PriorityChansConverger@ + . + The realization probably isn't + very fast and isn't good at memory economy, since it uses fresh high + level primitive - STM. It wasn't intended to be used + with millions of channels. But it's max throughput comparing to the + ordinary @Chan@ throughput is to be estimated (will do it in some + future version). +Version: 0.1 +category: Concurrency +Author: Andrejs Sisojevs +Maintainer: Andrejs Sisojevs +Copyright: Copyright (c) 2009-2010 Andrejs Sisojevs +License: LGPL +License-file: COPYRIGHT +Stability: experimental +Tested-with: GHC ==6.10.4 +Cabal-version: >=1.6 +Build-type: Simple +Extra-source-files: + COPYING + COPYRIGHT + doinst.sh + examples/SimplePCCTestEnvironment/SimplePCCTestEnvironment.hs +Build-depends: base ==4.*, containers -any, stm -any +Exposed-modules: + Control.Concurrent.ConcurrentUISupport + Control.Concurrent.STM.TChan.ExtsCommons + Control.Concurrent.STM.TChan.TChanL + Control.Concurrent.STM.TChan.TChanL_ + Control.Concurrent.STM.TChan.TChanB + Control.Concurrent.STM.TChan.TChanB_ + Control.Concurrent.PriorityChansConverger + Control.Concurrent.PriorityChansConverger.Commons + Control.Concurrent.PriorityChansConverger.PriorityChansConverger + Control.Concurrent.PriorityChansConverger.PriorityChansConvergerSTM +Other-modules: + Data.MyHelpers + Data.MyHelpers_ diff --git a/Setup.hs b/Setup.hs new file mode 100644 --- /dev/null +++ b/Setup.hs @@ -0,0 +1,4 @@ +#!/usr/bin/env runghc + +import Distribution.Simple +main = defaultMain diff --git a/doinst.sh b/doinst.sh new file mode 100644 --- /dev/null +++ b/doinst.sh @@ -0,0 +1,8 @@ +#! /bin/sh +clear + +sudo runghc Setup unregister +sudo runghc Setup configure --user +sudo runghc Setup build +sudo runghc Setup haddock +sudo runghc Setup install diff --git a/examples/SimplePCCTestEnvironment/SimplePCCTestEnvironment.hs b/examples/SimplePCCTestEnvironment/SimplePCCTestEnvironment.hs new file mode 100644 --- /dev/null +++ b/examples/SimplePCCTestEnvironment/SimplePCCTestEnvironment.hs @@ -0,0 +1,442 @@ +{- +Copyright (C) 2009-2010 Andrejs Sisojevs + +All rights reserved. + +For license and copyright information, see the file COPYRIGHT +-} + +-------------------------------------------------------------------------- +-------------------------------------------------------------------------- + +{- +abbr. PCC = PriorityChansConverger + +Table of content: + +1. Let's define system structure first. +1.1. Sources of input to our PCC. +1.2. Output of our PCC reader. +1.3. Put together PCC, it's input sources and it's output reader. + +2. Now, when the infrastructure is ready let's define some test cases and run them. +2.1. Test cases set. Not a complete set. If anybody wants to test it seriously, that would be great. +2.2. Running tests. + +3. Representations (instances of Show class). + +Appendix: Some helper functions. + +------------------------------------------- +Direct PCC usage is to be found only in parts 1.1., 1.2. and 1.3. + +-} + +{-# LANGUAGE ScopedTypeVariables, BangPatterns #-} + +module SimplePCCTestEnvironment where + +import Control.Concurrent +import Control.Concurrent.MVar +import Control.Concurrent.ConcurrentUISupport -- this module is provided by PCC package +import Control.Concurrent.STM +import Control.Concurrent.PriorityChansConverger -- this module is provided by PCC package +import Control.Monad +import qualified Control.Exception as E +import Data.List +import qualified Data.Map as M +import Data.Map (Map) +import qualified Safe.Failure as SF +import System.IO +import System.IO.Unsafe + +__useDumps = True + +---------------------------------------------------------------------- +---------------------------------------------------------------------- +-- 1. Let's define system structure first. +---------------------------------------------------------------------- +-- 1.1. Sources of input to our PCC. + +type PCCFillerID = Int -- PCC requires anything, that is of class Ord +type PCCFillerState = Int +type PCCFillerOutput = PCCFillerState +type PCCFillerCloser = ThreadFinisher +type MicrosecondsForDelay = Int +type SecondsForDelay = Int +data PCCFiller = + PCCFiller { + pccfInputID :: PCCFillerID -- local type + , pccfStartPriority :: StartPriority -- Int + , pccfChanCapacity :: ChanCapacity -- Int + , pccfPermitToBlock :: PermitToBlock -- Bool + , pccfDelayBetweenIterations :: MicrosecondsForDelay + , pccfStateMV :: MVar PCCFillerState + , pccfStateTransition :: PCCFillerState -> PCCFillerState + , pccfStopFiller :: MVar PCCFillerCloser + } + +defaultPCCFiller :: IO PCCFiller +defaultPCCFiller = do + state_mv <- newMVar 1 + stopper <- newMVar $ return () + return PCCFiller { + pccfInputID = undefined + , pccfStartPriority = undefined + , pccfChanCapacity = undefined + , pccfPermitToBlock = undefined + , pccfDelayBetweenIterations = undefined + , pccfStateMV = state_mv + , pccfStateTransition = (+ 1) + , pccfStopFiller = stopper + } + +runPCCFiller :: PCCFiller -> PriorityChansConverger PCCFillerID PCCFillerOutput -> ConcurrentUISupport -> IO () +runPCCFiller pccf pcc cuis = do + when __useDumps $ dump "" ("PCCFiller" ++ show channelID) WriteMode + interrupter_tv <- atomically $ newTVar False + cuisWrite cuis ("PCC filler (ID: " ++ show channelID ++ ") started.") + modifyMVar_ + (pccfStopFiller pccf) + (\ _ -> return $ do + cuisWrite cuis ("Signal to finish PCC filler (ID: " ++ show channelID ++ ") is sent.") + atomically $ modifyTVar_ interrupter_tv (const True) + ) + _ <- forkIO $ threadCycle interrupter_tv + return () + where + channelID = pccfInputID pccf + ------------ + threadCycle :: TVar InterruptShouldWe -> IO () + threadCycle interrupter_tv = _cycle + where + _cycle :: IO () -- It's possible to make this thread more STMish, use Control.Concurrent.PriorityChansConverger.PriorityChansConvergerSTM directly. + _cycle = do + stop <- atomically $ readTVar interrupter_tv + case stop of + True -> do + cuisWrite cuis ("PCC filler (ID: " ++ show channelID ++ ") finished.") + return () -- exit + False -> do + state <- modifyMVar + (pccfStateMV pccf) + (\ current_state -> + return (pccfStateTransition pccf current_state, current_state) + ) + when __useDumps $ reportExceptionIfAny_2 cuis "runPCCFiller:dump1" $ dump2 (return "\n----Before-write-----------------------------\n") ("PCCFiller" ++ show (pccfInputID pccf)) AppendMode + when __useDumps $ reportExceptionIfAny_2 cuis "runPCCFiller:dump2" $ dump2 (showPCC pcc) ("PCCFiller" ++ show (pccfInputID pccf)) AppendMode -- ######################################### + br_or_mb_failure <- case pccfPermitToBlock pccf of + True -> stmInterruptableWriteInPCC (readTVar interrupter_tv) channelID state pcc + False -> Right `liftM` writeInPCC False channelID state pcc + when __useDumps $ reportExceptionIfAny_2 cuis "runPCCFiller:dump3" $ dump2 (return "\n----After-write------------------------------\n") ("PCCFiller" ++ show (pccfInputID pccf)) AppendMode + when __useDumps $ reportExceptionIfAny_2 cuis "runPCCFiller:dump4" $ dump2 (showPCC pcc) ("PCCFiller" ++ show (pccfInputID pccf)) AppendMode -- ######################################### + case br_or_mb_failure of + Left () -> cuisWrite cuis ("Writing to channel (ID: " ++ show channelID ++ ") was interrupted." ) + Right Nothing -> cuisWrite cuis ("Wrote to channel (ID: " ++ show channelID ++ "): " ++ show (channelID, state) ) + Right (Just BadKey_FRWPCC) -> cuisWrite cuis ("Write to channel (ID: " ++ show channelID ++ ") FAILED: bad key." ) + Right (Just ChanFull_FRWPCC) -> cuisWrite cuis ("Write to channel (ID: " ++ show channelID ++ ") FAILED: not enough space in channel." ) + stop2 <- atomically $ readTVar interrupter_tv + unless stop2 $ threadDelay (pccfDelayBetweenIterations pccf) + _cycle + +runPCCFillersArray :: [PCCFiller] -> PriorityChansConverger PCCFillerID PCCFillerOutput -> ConcurrentUISupport -> IO () +runPCCFillersArray pccf_list pcc cuis = mapM_ (\ pccf -> runPCCFiller pccf pcc cuis) pccf_list + +stopPCCFiller :: PCCFiller -> IO () +stopPCCFiller = join . readMVar . pccfStopFiller + +stopPCCFillersArray :: [PCCFiller] -> IO () +stopPCCFillersArray = mapM_ stopPCCFiller + +---------------------------------------------------------------------- +-- 1.2. Output of our PCC reader. +---------------------------------------------------------------------- + +type PCCReaderID = Int + +data PCCReader = + PCCReader { + pccrReaderID :: PCCReaderID + , pccrPermitToBlock :: PermitToBlock + , pccrMicrosecondsForDelay :: MicrosecondsForDelay + , pccrStopFiller :: MVar PCCFillerCloser + } + +defaultPCCReader :: IO PCCReader +defaultPCCReader = do + stopper <- newMVar $ return () + return PCCReader { + pccrReaderID = undefined + , pccrPermitToBlock = undefined + , pccrMicrosecondsForDelay = undefined + , pccrStopFiller = stopper + } + +runPCCReader :: PCCReader -> PriorityChansConverger PCCFillerID PCCFillerOutput -> ConcurrentUISupport -> IO () +runPCCReader pccr pcc cuis = do + when __useDumps $ dump "" f_name WriteMode + interrupter_tv <- atomically $ newTVar False + cuisWrite cuis ("PCC reader (ID: " ++ show (pccrReaderID pccr) ++ ") started.") + _ <- forkIO $ threadCycle interrupter_tv + modifyMVar_ + (pccrStopFiller pccr) + (\ _ -> return $ do + cuisWrite cuis ("Signal to finish PCC reader (ID: " ++ show (pccrReaderID pccr) ++ ") is sent.") + atomically $ modifyTVar_ interrupter_tv (const True) + ) + return () + where + f_name = "PCCReader" ++ show (pccrReaderID pccr) + ------------ + threadCycle :: TVar InterruptShouldWe -> IO () + threadCycle interrupter_tv = _cycle + where + _cycle :: IO () -- It's possible to make this thread more STMish, use Control.Concurrent.PriorityChansConverger.PriorityChansConvergerSTM directly. + _cycle = do + stop <- atomically $ readTVar interrupter_tv + case stop of + True -> do + cuisWrite cuis ("PCC reader (ID: " ++ show (pccrReaderID pccr) ++ ") finished.") + return () -- exit + False -> do + when __useDumps $ reportExceptionIfAny_2 cuis "runPCCReader:dump1" $ dump2 (return "\n----Before-read------------------------------\n") f_name AppendMode + when __useDumps $ reportExceptionIfAny_2 cuis "runPCCReader:dump2" $ dump2 (showPCC pcc) f_name AppendMode -- ######################################### + br_or_elem <- case pccrPermitToBlock pccr of + True -> stmInterruptableReadFromPCC (readTVar interrupter_tv) pcc + False -> do + mb_r <- readFromPCC False pcc + return $ case mb_r of + Nothing -> Left () + Just r -> Right r + when __useDumps $ reportExceptionIfAny_2 cuis "runPCCReader:dump3" $ dump2 (return "\n----After-read-------------------------------\n") f_name AppendMode + when __useDumps $ reportExceptionIfAny_2 cuis "runPCCReader:dump4" $ dump2 (showPCC pcc) f_name AppendMode -- ######################################### + case (pccrPermitToBlock pccr, br_or_elem) of + (True , Left ()) -> cuisWrite cuis ("Read from PCC (reader ID: " ++ show (pccrReaderID pccr) ++ ") is interrupred.") + (False, Left ()) -> cuisWrite cuis ("Read from PCC (reader ID: " ++ show (pccrReaderID pccr) ++ ") failed: PCC is empty.") + (_ , Right (k, e)) -> cuisWrite cuis ("Read from PCC (reader ID: " ++ show (pccrReaderID pccr) ++ "): " ++ show (k, e)) + stop2 <- atomically $ readTVar interrupter_tv + unless stop2 $ threadDelay (pccrMicrosecondsForDelay pccr) + _cycle + +stopPCCReader :: PCCReader -> IO () +stopPCCReader = join . readMVar . pccrStopFiller + +runPCCReadersArray :: [PCCReader] -> PriorityChansConverger PCCFillerID PCCFillerOutput -> ConcurrentUISupport -> IO () +runPCCReadersArray pccr_list pcc cuis = mapM_ (\ pccr -> runPCCReader pccr pcc cuis) pccr_list + +stopPCCReadersArray :: [PCCReader] -> IO () +stopPCCReadersArray = mapM_ stopPCCReader + +---------------------------------------------------------------------- +-- 1.3. Put together PCC, it's input sources and it's output reader. +---------------------------------------------------------------------- + +listOfFillers_to_ChansMapToConstructPCC :: [PCCFiller] -> Map PCCFillerID (StartPriority, ChanCapacity) +listOfFillers_to_ChansMapToConstructPCC pccf_list = + M.fromList $ map + (\ pccf -> + ( pccfInputID pccf + , (pccfStartPriority pccf + , pccfChanCapacity pccf + ) ) + ) + pccf_list + +constructPCC :: [PCCFiller] -> ConcurrentUISupport -> IO (PriorityChansConverger PCCFillerID PCCFillerOutput) +constructPCC pccf_list cuis = newPriorityChansConverger_wCUIS (Just cuis) $ listOfFillers_to_ChansMapToConstructPCC pccf_list +-- When constructing new PCC, it is possible to use 'newPriorityChansConverger' +-- instead of 'newPriorityChansConverger_wCUIS' which isn't using +-- any 'ConcurrentUISupport' + +runFillersAndPCCAndReaders :: [PCCFiller] -> [PCCReader] -> ConcurrentUISupport -> IO ThreadFinisher +runFillersAndPCCAndReaders pccf_list pccr_list cuis = do + pcc <- constructPCC pccf_list cuis + runPCCFillersArray pccf_list pcc cuis + runPCCReadersArray pccr_list pcc cuis + interrupter_tv <- atomically $ newTVar False + _ <- forkIO $ do + atomically $ do + stop <- readTVar interrupter_tv + case stop of + True -> return () + False -> retry + stopPCCFillersArray pccf_list + stopPCCReadersArray pccr_list + return (atomically $ writeTVar interrupter_tv True) + +---------------------------------------------------------------------- +---------------------------------------------------------------------- +-- 2. Now, when the infrastructure is ready let's define +-- some test cases and run them +---------------------------------------------------------------------- +---------------------------------------------------------------------- +-- 2.1. Test cases set. +-- Not a complete set. If anybody wants to test it seriously, +-- that would be great. + +type InputMode = Int +inputChansSet :: InputMode -> IO [PCCFiller] +inputChansSet 1 = sequence [] +inputChansSet 2 = sequence [ defaultPCCFiller >>= \ pccf -> return pccf { pccfInputID = 1, pccfStartPriority = 10, pccfChanCapacity = 0, pccfPermitToBlock = False, pccfDelayBetweenIterations = 1000000 } ] +inputChansSet 3 = sequence [ defaultPCCFiller >>= \ pccf -> return pccf { pccfInputID = 1, pccfStartPriority = 10, pccfChanCapacity = 1, pccfPermitToBlock = True, pccfDelayBetweenIterations = 1000000 } ] +inputChansSet 4 = sequence [ defaultPCCFiller >>= \ pccf -> return pccf { pccfInputID = 1, pccfStartPriority = 5, pccfChanCapacity = 3, pccfPermitToBlock = False, pccfDelayBetweenIterations = 900000 } + , defaultPCCFiller >>= \ pccf -> return pccf { pccfInputID = 2, pccfStartPriority = 3, pccfChanCapacity = 5, pccfPermitToBlock = False, pccfDelayBetweenIterations = 1200000 } + ] +inputChansSet 5 = sequence [ defaultPCCFiller >>= \ pccf -> return pccf { pccfInputID = 1, pccfStartPriority = 10, pccfChanCapacity = 4, pccfPermitToBlock = True, pccfDelayBetweenIterations = 1000000 } + , defaultPCCFiller >>= \ pccf -> return pccf { pccfInputID = 2, pccfStartPriority = 8, pccfChanCapacity = 6, pccfPermitToBlock = True, pccfDelayBetweenIterations = 1000000 } + , defaultPCCFiller >>= \ pccf -> return pccf { pccfInputID = 3, pccfStartPriority = 6, pccfChanCapacity = 8, pccfPermitToBlock = True, pccfDelayBetweenIterations = 1000000 } + , defaultPCCFiller >>= \ pccf -> return pccf { pccfInputID = 4, pccfStartPriority = 4, pccfChanCapacity = 10, pccfPermitToBlock = True, pccfDelayBetweenIterations = 1000000 } + ] +inputChansSet 6 = sequence [ defaultPCCFiller >>= \ pccf -> return pccf { pccfInputID = 1, pccfStartPriority = 10, pccfChanCapacity = 4, pccfPermitToBlock = True, pccfDelayBetweenIterations = 200000 } + , defaultPCCFiller >>= \ pccf -> return pccf { pccfInputID = 2, pccfStartPriority = 8, pccfChanCapacity = 6, pccfPermitToBlock = False, pccfDelayBetweenIterations = 200000 } + , defaultPCCFiller >>= \ pccf -> return pccf { pccfInputID = 3, pccfStartPriority = 6, pccfChanCapacity = 8, pccfPermitToBlock = False, pccfDelayBetweenIterations = 1000000 } + , defaultPCCFiller >>= \ pccf -> return pccf { pccfInputID = 4, pccfStartPriority = 4, pccfChanCapacity = 10, pccfPermitToBlock = True, pccfDelayBetweenIterations = 1000000 } + ] +__inputChansSet_lowerBound = 1 +__inputChansSet_higherBound = 6 + +type OutputMode = Int +readerSet :: OutputMode -> IO [PCCReader] +readerSet 1 = sequence [ defaultPCCReader >>= \ pccr -> return pccr { pccrReaderID = 1, pccrPermitToBlock = False, pccrMicrosecondsForDelay = 1000000 } ] +readerSet 2 = sequence [ defaultPCCReader >>= \ pccr -> return pccr { pccrReaderID = 1, pccrPermitToBlock = True, pccrMicrosecondsForDelay = 1000000 } ] +readerSet 3 = sequence [ defaultPCCReader >>= \ pccr -> return pccr { pccrReaderID = 1, pccrPermitToBlock = False, pccrMicrosecondsForDelay = 500000 } ] +readerSet 4 = sequence [ defaultPCCReader >>= \ pccr -> return pccr { pccrReaderID = 1, pccrPermitToBlock = True, pccrMicrosecondsForDelay = 500000 } ] +readerSet 5 = sequence [ defaultPCCReader >>= \ pccr -> return pccr { pccrReaderID = 1, pccrPermitToBlock = False, pccrMicrosecondsForDelay = 1500000 } ] +readerSet 6 = sequence [ defaultPCCReader >>= \ pccr -> return pccr { pccrReaderID = 1, pccrPermitToBlock = True, pccrMicrosecondsForDelay = 1500000 } ] +readerSet 7 = sequence [ defaultPCCReader >>= \ pccr -> return pccr { pccrReaderID = 1, pccrPermitToBlock = False, pccrMicrosecondsForDelay = 500000 } + , defaultPCCReader >>= \ pccr -> return pccr { pccrReaderID = 2, pccrPermitToBlock = True, pccrMicrosecondsForDelay = 500000 } + ] +readerSet 8 = sequence [ defaultPCCReader >>= \ pccr -> return pccr { pccrReaderID = 1, pccrPermitToBlock = True, pccrMicrosecondsForDelay = 200000 } + , defaultPCCReader >>= \ pccr -> return pccr { pccrReaderID = 2, pccrPermitToBlock = True, pccrMicrosecondsForDelay = 1500000 } + ] +readerSet 9 = sequence [ defaultPCCReader >>= \ pccr -> return pccr { pccrReaderID = 1, pccrPermitToBlock = True, pccrMicrosecondsForDelay = 150000 } + , defaultPCCReader >>= \ pccr -> return pccr { pccrReaderID = 2, pccrPermitToBlock = False, pccrMicrosecondsForDelay = 200000 } + ] +__readerConfig_lowerBound = 1 +__readerConfig_higherBound = 9 + +---------------------------------------------------------------------- +-- 2.2. Running tests. +---------------------------------------------------------------------- + +type RunTest = Bool +runTest :: (RunTest, InputMode, OutputMode, SecondsForDelay) -> ConcurrentUISupport -> IO () +runTest (test_shouldwe, input_mode, output_mode, run_time) cuis = + E.handle + (\ se -> print (se :: E.SomeException)) + $ do pccr_list <- readerSet output_mode + pccf_list <- inputChansSet input_mode + case test_shouldwe of + False -> + let str = "" + ++ "PCCFillers(" ++ show (length pccf_list) ++ "): \n" + ++ (join $ intersperse "\n" $ map show pccf_list) + ++ "\n-----------------------------------\n" + ++ "PCCReaders(" ++ show (length pccf_list) ++ "): \n" + ++ (join $ intersperse "\n" $ map show pccr_list) + in cuisWrite cuis str + True -> do + finisher <- runFillersAndPCCAndReaders pccf_list pccr_list cuis + threadDelay (run_time * 1000000) + finisher + +main = do + cuis <- defaultConcurrentUISupportIO_logInFile "PCCTest.log" WriteMode -- provided by the PCC package, Control.Concurrent.SimpleUI + runConcurrentUISupport cuis + cuisWrite cuis ("Welcome to SimplePCCTestEnvironment v0.1!\n ") + _cycle cuis + cuisFinish cuis + where + _cycle :: ConcurrentUISupport -> IO () + _cycle cuis = do + cuisWrite cuis ("Please input mode (True/False, inputChansSet #{" ++ show __inputChansSet_lowerBound ++ ".." ++ show __inputChansSet_higherBound ++ "}, readerSet #{" ++ show __readerConfig_lowerBound ++ ".." ++ show __readerConfig_higherBound ++ "}, time (in seconds) for how long to run test). ") + cuisWrite cuis ("Or type ':q' for exit.") + cuisWrite cuis ("If first is True, then test is run, otherwise chosen inputChansSet and readerSet are shown.") + cuisWrite cuis ("Second parameter should be index of inputChansSet, third - of readerSet.") + cuisWrite cuis ("Example inputs: (True, 1, 1, 20) or (False,1,2,10), but not (1,2,3,4) nor (True," ++ show (__inputChansSet_lowerBound - 1) ++ "," ++ show (__readerConfig_higherBound + 1) ++ ",-1) nor (True,2,3,4,5)") + cuisWrite cuis ("Input > ") + mode <- cuisReadLn cuis "main thread" -- (True, 5, 9, 20) -- (True, 6, 6, 30) + case mode == ":q" of + True -> return () + False -> do + err_or_mode <- (E.try $ SF.read mode) :: IO (Either E.SomeException (RunTest, InputMode, OutputMode, SecondsForDelay)) + case err_or_mode of + Left se -> cuisWrite cuis (show se) >> cuisNewPage cuis + Right tm@(test_shouldwe, input_mode, output_mode, run_time) -> + let cond_1 = input_mode `betweenI` (__inputChansSet_lowerBound, __inputChansSet_higherBound) + cond_2 = output_mode `betweenI` ( __readerConfig_lowerBound, __readerConfig_higherBound) + cond_3 = run_time > 0 + cond = cond_1 && cond_2 && cond_3 + in case cond of + False -> return () + True -> runTest tm cuis >> threadDelay 1000000 >> cuisNewPage cuis + _cycle cuis + +---------------------------------------------------------------------- +---------------------------------------------------------------------- +-- 3. Representations (instances of Show class). +---------------------------------------------------------------------- +---------------------------------------------------------------------- + +instance Show PCCFiller where + show pccf = + "PCCFiller {" + ++ "\n ID = " ++ (show $ pccfInputID pccf) + ++ "\n StartPriority = " ++ (show $ pccfInputID pccf) + ++ "\n ChanCapacity = " ++ (show $ pccfChanCapacity pccf) + ++ "\n PermitToBlock = " ++ (show $ pccfPermitToBlock pccf) + ++ "\n MicrosecondsForDelay = " ++ (show $ pccfDelayBetweenIterations pccf) + ++ "\n PCCFillerState = " ++ (show $ unsafePerformIO $ readMVar $ pccfStateMV pccf) + ++ "\n}" + +instance Show PCCReader where + show pccr = + "PCCReader {" + ++ "\n ID = " ++ (show $ pccrReaderID pccr) + ++ "\n MicrosecondsForDelay = " ++ (show $ pccrMicrosecondsForDelay pccr) + ++ "\n PermitToBlock = " ++ (show $ pccrPermitToBlock pccr) + ++ "}" + +---------------------------------------------------------------------- +-- Appendix: Some helper functions. +---------------------------------------------------------------------- + +infixr 1 << +(<<) :: Monad m => m b -> m a -> m b +f << x = x >> f + +betweenI :: Ord a => a -> (a, a) -> Bool +betweenNI :: Ord a => a -> (a, a) -> Bool +betweenI a (l,h) = a >= l && a <= h +betweenNI a (l,h) = a > l && a < h + +infixr 1 `countIn` +countIn :: Eq a => a -> [a] -> Int +countIn a list = _count list 0 + where + _count [] !i = i + _count (h:t) !i = + case h == a of + True -> _count t (i + 1) + False -> _count t i + +modifyTVar_ :: TVar a -> (a -> a) -> STM () +modifyTVar_ tv f = do + v <- readTVar tv + writeTVar tv (f v) + +-- data System.IO.IOMode = ReadMode | WriteMode | AppendMode | ReadWriteMode +type WhatToDump = String +type FileSuffix = String +dump :: WhatToDump -> FileSuffix -> IOMode -> IO () +dump a s m = do + h <- openFile ("./dump.out." ++ s ++ ".hs") m + hPutStr h $ show a + hClose h + +dump2 :: IO WhatToDump -> FileSuffix -> IOMode -> IO () +dump2 a s m = do + h <- openFile ("./dump.out." ++ s ++ ".hs") m + a >>= hPutStr h + hClose h + --a >>= putStrLn